v1-0003-Custom-compression-methods.patch

From 243f62308c18e6ca79d74244d00f8a81d512fc5d Mon Sep 17 00:00:00 2001
From: Dilip Kumar <dilipkumar@localhost.localdomain>
Date: Wed, 12 Aug 2020 14:35:38 +0530
Subject: [PATCH v1 3/3] Custom compression methods

Allow to create custom compression methods.  This also allow to change
the compression methods for the existing column.
---
 doc/src/sgml/catalogs.sgml                  |  16 +-
 doc/src/sgml/compression-am.sgml            | 178 ++++++
 doc/src/sgml/filelist.sgml                  |   1 +
 doc/src/sgml/indexam.sgml                   |   2 +-
 doc/src/sgml/postgres.sgml                  |   1 +
 doc/src/sgml/ref/alter_table.sgml           |  18 +
 doc/src/sgml/ref/create_access_method.sgml  |   3 +-
 doc/src/sgml/ref/create_table.sgml          |  10 +-
 doc/src/sgml/storage.sgml                   |   4 +-
 src/backend/access/common/detoast.c         |  21 +-
 src/backend/access/common/reloptions.c      |  88 +++
 src/backend/access/common/toast_internals.c | 167 +++++-
 src/backend/access/compression/cm_pglz.c    |  71 ++-
 src/backend/access/compression/cm_zlib.c    |  89 ++-
 src/backend/access/compression/cmapi.c      |  53 ++
 src/backend/access/heap/heapam.c            |  16 +-
 src/backend/access/heap/heaptoast.c         |   6 +-
 src/backend/access/heap/rewriteheap.c       |   2 +-
 src/backend/access/table/toast_helper.c     |  78 ++-
 src/backend/catalog/Makefile                |   2 +-
 src/backend/catalog/dependency.c            |   9 +
 src/backend/catalog/objectaddress.c         |  58 ++
 src/backend/commands/alter.c                |   1 +
 src/backend/commands/compressioncmds.c      | 623 +++++++++++++++++++-
 src/backend/commands/copy.c                 |   4 +-
 src/backend/commands/createas.c             |   2 +-
 src/backend/commands/event_trigger.c        |   1 +
 src/backend/commands/foreigncmds.c          |  46 +-
 src/backend/commands/matview.c              |   2 +-
 src/backend/commands/tablecmds.c            | 369 +++++++++++-
 src/backend/nodes/copyfuncs.c               |  17 +-
 src/backend/nodes/equalfuncs.c              |  15 +-
 src/backend/nodes/nodeFuncs.c               |   8 +
 src/backend/nodes/outfuncs.c                |  15 +-
 src/backend/parser/gram.y                   |  57 +-
 src/backend/parser/parse_utilcmd.c          |   2 +-
 src/backend/utils/adt/pg_upgrade_support.c  |  10 +
 src/backend/utils/cache/syscache.c          |  12 +
 src/bin/pg_dump/pg_dump.c                   | 148 ++++-
 src/bin/pg_dump/pg_dump.h                   |  15 +
 src/bin/pg_dump/t/002_pg_dump.pl            |  95 +++
 src/bin/psql/describe.c                     |  11 +-
 src/include/access/cmapi.h                  |  13 +-
 src/include/access/heapam.h                 |   3 +-
 src/include/access/heaptoast.h              |   3 +-
 src/include/access/hio.h                    |   2 +
 src/include/access/toast_helper.h           |   4 +-
 src/include/access/toast_internals.h        |  21 +-
 src/include/catalog/binary_upgrade.h        |   2 +
 src/include/catalog/dependency.h            |   5 +-
 src/include/catalog/indexing.h              |   5 +
 src/include/catalog/pg_attr_compression.dat |  24 +
 src/include/catalog/pg_attr_compression.h   |  53 ++
 src/include/catalog/pg_proc.dat             |   7 +
 src/include/catalog/toasting.h              |   1 +
 src/include/commands/defrem.h               |  12 +-
 src/include/nodes/nodes.h                   |   1 +
 src/include/nodes/parsenodes.h              |  19 +-
 src/include/utils/syscache.h                |   1 +
 src/test/regress/expected/create_cm.out     | 376 +++++++++++-
 src/test/regress/expected/create_cm_1.out   | 407 +++++++++++++
 src/test/regress/expected/create_index.out  |   6 +-
 src/test/regress/expected/psql.out          |  12 +-
 src/test/regress/expected/sanity_check.out  |   3 +
 src/test/regress/sql/create_cm.sql          | 171 +++++-
 src/tools/pgindent/typedefs.list            |   1 +
 66 files changed, 3277 insertions(+), 221 deletions(-)
 create mode 100644 doc/src/sgml/compression-am.sgml
 create mode 100644 src/include/catalog/pg_attr_compression.dat
 create mode 100644 src/include/catalog/pg_attr_compression.h
 create mode 100644 src/test/regress/expected/create_cm_1.out

diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 5ef5b69e81..642a182f2a 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -70,6 +70,11 @@
       <entry>access method support functions</entry>
      </row>
 
+     <row>
+      <entry><link linkend="catalog-pg-attr-compression"><structname>pg_attr_compression</structname></link></entry>
+      <entry>table columns compression relationships and options</entry>
+     </row>
+
      <row>
       <entry><link linkend="catalog-pg-attrdef"><structname>pg_attrdef</structname></link></entry>
       <entry>column default values</entry>
@@ -660,8 +665,9 @@
    relation access methods.  There is one row for each access method supported
    by the system.
    Currently, there are tables, indexes and compression access methods.
-   The requirements for table and index access methods are discussed in detail
-   in <xref linkend="tableam"/> and <xref linkend="indexam"/> respectively.
+   The requirements for table, index and compression access methods
+   are discussed in detail in <xref linkend="tableam"/>,
+   <xref linkend="indexam"/> and <xref linkend="compression-am"/> respectively.
   </para>
 
   <table>
@@ -1011,6 +1017,12 @@
 
  </sect1>
 
+ <sect1 id="catalog-pg-attr-compression">
+  <title><structname>pg_attr_compression</structname></title>
+  <indexterm zone="catalog-pg-attr-compression">
+   <primary>pg_attr_compression</primary>
+  </indexterm>
+ </sect1>
 
  <sect1 id="catalog-pg-attrdef">
   <title><structname>pg_attrdef</structname></title>
diff --git a/doc/src/sgml/compression-am.sgml b/doc/src/sgml/compression-am.sgml
new file mode 100644
index 0000000000..e23d817910
--- /dev/null
+++ b/doc/src/sgml/compression-am.sgml
@@ -0,0 +1,178 @@
+<!-- doc/src/sgml/compression-am.sgml -->
+
+<chapter id="compression-am">
+ <title>Compression Access Methods</title>
+  <para>
+   <productname>PostgreSQL</productname> supports two internal
+   built-in compression methods (<literal>pglz</literal>
+   and <literal>zlib</literal>), and also allows to add more custom compression
+   methods through compression access methods interface.
+  </para>
+
+ <sect1 id="builtin-compression-methods">
+  <title>Built-in Compression Access Methods</title>
+  <para>
+   These compression access methods are included in
+   <productname>PostgreSQL</productname> and don't need any external extensions.
+  </para>
+  <table id="builtin-compression-methods-table">
+   <title>Built-in Compression Access Methods</title>
+   <tgroup cols="2">
+    <thead>
+     <row>
+      <entry>Name</entry>
+      <entry>Options</entry>
+     </row>
+    </thead>
+    <tbody>
+     <row>
+      <entry><literal>pglz</literal></entry>
+      <entry>
+       <literal>min_input_size (int)</literal>,
+       <literal>max_input_size (int)</literal>,
+       <literal>min_comp_rate (int)</literal>,
+       <literal>first_success_by (int)</literal>,
+       <literal>match_size_good (int)</literal>,
+       <literal>match_size_drop (int)</literal>
+      </entry>
+     </row>
+     <row>
+      <entry><literal>zlib</literal></entry>
+      <entry><literal>level (text)</literal>, <literal>dict (text)</literal></entry>
+     </row>
+    </tbody>
+   </tgroup>
+  </table>
+  <para>
+  Note that for <literal>zlib</literal> to work it should be installed in the
+  system and <productname>PostgreSQL</productname> should be compiled without
+  <literal>--without-zlib</literal> flag.
+  </para>
+ </sect1>
+
+ <sect1 id="compression-api">
+  <title>Basic API for compression methods</title>
+
+  <para>
+   Each compression access method is described by a row in the
+   <link linkend="catalog-pg-am"><structname>pg_am</structname></link>
+   system catalog.  The <structname>pg_am</structname> entry
+   specifies a name and a <firstterm>handler function</firstterm> for the access
+   method.  These entries can be created and deleted using the
+   <xref linkend="sql-create-access-method"/> and
+   <xref linkend="sql-drop-access-method"/> SQL commands.
+  </para>
+
+  <para>
+   A compression access method handler function must be declared to accept a
+   single argument of type <type>internal</type> and to return the
+   pseudo-type <type>compression_am_handler</type>.  The argument is a dummy value that
+   simply serves to prevent handler functions from being called directly from
+   SQL commands.  The result of the function must be a palloc'd struct of
+   type <structname>CompressionAmRoutine</structname>, which contains everything
+   that the core code needs to know to make use of the compression access method.
+   The <structname>CompressionAmRoutine</structname> struct, also called the access
+   method's <firstterm>API struct</firstterm>, contains pointers to support
+   functions for the access method. These support functions are plain C
+   functions and are not visible or callable at the SQL level.
+   The support functions are described in <xref linkend="compression-am-functions"/>.
+  </para>
+
+  <para>
+   The structure <structname>CompressionAmRoutine</structname> is defined thus:
+<programlisting>
+typedef struct CompressionAmRoutine
+{
+    NodeTag     type;
+
+    cmcheck_function        cmcheck;        /* can be NULL */
+    cminitstate_function    cminitstate;    /* can be NULL */
+    cmcompress_function     cmcompress;
+    cmcompress_function     cmdecompress;
+} CompressionAmRoutine;
+</programlisting>
+  </para>
+ </sect1>
+ <sect1 id="compression-am-functions">
+  <title>Compression Access Method Functions</title>
+
+  <para>
+   The compression and auxiliary functions that an compression access
+   method must provide in <structname>CompressionAmRoutine</structname> are:
+  </para>
+
+  <para>
+<programlisting>
+void
+cmcheck (Form_pg_attribute att, List *options);
+</programlisting>
+   Called when an attribute is linked with compression access method. Could
+   be used to check compatibility with the attribute and other additional
+   checks.
+  </para>
+
+  <para>
+  Compression functions take special struct
+  <structname>CompressionAmOptions</structname> as first
+  parameter. This struct contains per backend cached state for each
+  attribute compression record. CompressionAmOptions is defined thus:
+
+<programlisting>
+typedef struct CompressionAmOptions
+{
+    Oid         acoid;          /* Oid of attribute compression */
+    Oid         amoid;          /* Oid of compression access method */
+    List       *acoptions;      /* Parsed options, used for comparison */
+    CompressionAmRoutine *amroutine;    /* compression access method routine */
+
+    /* result of cminitstate function will be put here */
+    void       *acstate;
+} CompressionAmOptions;
+</programlisting>
+  </para>
+
+  <para>
+  The <structfield>acstate</structfield> field is used to keep temporary state
+  between compression functions calls and stores the result of
+  <structfield>cminitstate</structfield> function. It could be useful to store
+  the parsed view of the compression options.
+  </para>
+
+  <para>
+  Note that any invalidation of <structname>pg_attr_compression</structname> relation
+  will cause all the cached <structfield>acstate</structfield> options cleared.
+  They will be recreated on the next compression functions calls.
+  </para>
+
+  <para>
+<programlisting>
+void *
+cminitstate (Oid acoid, List *options);
+</programlisting>
+  Called when <structname>CompressionAmOptions</structname> is being
+  initialized. Can return a pointer to memory that will be passed between
+  compression functions calls.
+  </para>
+
+  <para>
+<programlisting>
+struct varlena *
+cmcompress (CompressionAmOptions *cmoptions,
+            const struct varlena *value);
+</programlisting>
+   Function is used to compress varlena. Could return NULL if data is
+   incompressible. If it returns varlena bigger than original the core will
+   not use it.
+  </para>
+
+  <para>
+<programlisting>
+struct varlena *
+cmdecompress (CompressionAmOptions *cmoptions,
+              const struct varlena *value);
+</programlisting>
+   Function is used to decompress varlena.
+  </para>
+
+ </sect1>
+</chapter>
diff --git a/doc/src/sgml/filelist.sgml b/doc/src/sgml/filelist.sgml
index 64b5da0070..ec7f2b5ca9 100644
--- a/doc/src/sgml/filelist.sgml
+++ b/doc/src/sgml/filelist.sgml
@@ -91,6 +91,7 @@
 <!ENTITY planstats    SYSTEM "planstats.sgml">
 <!ENTITY tableam    SYSTEM "tableam.sgml">
 <!ENTITY indexam    SYSTEM "indexam.sgml">
+<!ENTITY compression-am SYSTEM "compression-am.sgml">
 <!ENTITY nls        SYSTEM "nls.sgml">
 <!ENTITY plhandler  SYSTEM "plhandler.sgml">
 <!ENTITY fdwhandler SYSTEM "fdwhandler.sgml">
diff --git a/doc/src/sgml/indexam.sgml b/doc/src/sgml/indexam.sgml
index 1aea4db707..8f63c4cac9 100644
--- a/doc/src/sgml/indexam.sgml
+++ b/doc/src/sgml/indexam.sgml
@@ -55,7 +55,7 @@
   <title>Basic API Structure for Indexes</title>
 
   <para>
-   Each index access method is described by a row in the
+   Each index access method is described by a row with INDEX type in the
    <link linkend="catalog-pg-am"><structname>pg_am</structname></link>
    system catalog.  The <structname>pg_am</structname> entry
    specifies a name and a <firstterm>handler function</firstterm> for the index
diff --git a/doc/src/sgml/postgres.sgml b/doc/src/sgml/postgres.sgml
index c41ce9499b..2d14e32775 100644
--- a/doc/src/sgml/postgres.sgml
+++ b/doc/src/sgml/postgres.sgml
@@ -259,6 +259,7 @@ break is not needed in a wider output rendering.
   &geqo;
   &tableam;
   &indexam;
+  &compression-am;
   &generic-wal;
   &btree;
   &gist;
diff --git a/doc/src/sgml/ref/alter_table.sgml b/doc/src/sgml/ref/alter_table.sgml
index b2eb7097a9..1668a007b5 100644
--- a/doc/src/sgml/ref/alter_table.sgml
+++ b/doc/src/sgml/ref/alter_table.sgml
@@ -54,6 +54,7 @@ ALTER TABLE [ IF EXISTS ] <replaceable class="parameter">name</replaceable>
     ALTER [ COLUMN ] <replaceable class="parameter">column_name</replaceable> SET ( <replaceable class="parameter">attribute_option</replaceable> = <replaceable class="parameter">value</replaceable> [, ... ] )
     ALTER [ COLUMN ] <replaceable class="parameter">column_name</replaceable> RESET ( <replaceable class="parameter">attribute_option</replaceable> [, ... ] )
     ALTER [ COLUMN ] <replaceable class="parameter">column_name</replaceable> SET STORAGE { PLAIN | EXTERNAL | EXTENDED | MAIN }
+    ALTER [ COLUMN ] <replaceable class="parameter">column_name</replaceable> SET COMPRESSION <replaceable class="parameter">compression_am</replaceable> [ WITH (<replaceable class="parameter">compression_am_options</replaceable>) ] [ PRESERVE (<replaceable class="parameter">compression_preserve_list</replaceable>) ]
     ADD <replaceable class="parameter">table_constraint</replaceable> [ NOT VALID ]
     ADD <replaceable class="parameter">table_constraint_using_index</replaceable>
     ALTER CONSTRAINT <replaceable class="parameter">constraint_name</replaceable> [ DEFERRABLE | NOT DEFERRABLE ] [ INITIALLY DEFERRED | INITIALLY IMMEDIATE ]
@@ -383,6 +384,23 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
     </listitem>
    </varlistentry>
 
+   <varlistentry>
+    <term>
+     <literal>SET COMPRESSION <replaceable class="parameter">compression_method_name</replaceable> [ WITH (<replaceable class="parameter">compression_method_options</replaceable>) ] [ PRESERVE (<replaceable class="parameter">compression_preserve_list</replaceable>) ]</literal>
+    </term>
+    <listitem>
+     <para>
+      This form adds compression to a column. Compression access method should be
+      created with <xref linkend="sql-create-access-method"/>. If compression
+      method has options they could be specified with <literal>WITH</literal>
+      parameter. The PRESERVE list contains list of compression access methods
+      used on the column and determines which of them should be kept on the
+      column. Without PRESERVE or partial list of compression methods table
+      will be rewritten.
+     </para>
+    </listitem>
+   </varlistentry>
+
    <varlistentry>
     <term><literal>ADD <replaceable class="parameter">table_constraint</replaceable> [ NOT VALID ]</literal></term>
     <listitem>
diff --git a/doc/src/sgml/ref/create_access_method.sgml b/doc/src/sgml/ref/create_access_method.sgml
index 3da27c8ce9..79f1290a58 100644
--- a/doc/src/sgml/ref/create_access_method.sgml
+++ b/doc/src/sgml/ref/create_access_method.sgml
@@ -84,7 +84,8 @@ CREATE ACCESS METHOD <replaceable class="parameter">name</replaceable>
       The C-level API that the handler function must implement varies
       depending on the type of access method. The table access method API
       is described in <xref linkend="tableam"/>, the index access method
-      API is described in <xref linkend="indexam"/>.
+      API is described in <xref linkend="indexam"/> an the compression access
+      method API is described in <xref linkend="compression-am"/>.
      </para>
     </listitem>
    </varlistentry>
diff --git a/doc/src/sgml/ref/create_table.sgml b/doc/src/sgml/ref/create_table.sgml
index 1a01c91b85..c47b184add 100644
--- a/doc/src/sgml/ref/create_table.sgml
+++ b/doc/src/sgml/ref/create_table.sgml
@@ -69,7 +69,7 @@ CREATE [ [ GLOBAL | LOCAL ] { TEMPORARY | TEMP } | UNLOGGED ] TABLE [ IF NOT EXI
   GENERATED { ALWAYS | BY DEFAULT } AS IDENTITY [ ( <replaceable>sequence_options</replaceable> ) ] |
   UNIQUE <replaceable class="parameter">index_parameters</replaceable> |
   PRIMARY KEY <replaceable class="parameter">index_parameters</replaceable> |
-  COMPRESSION <replaceable class="parameter">compression_access_method</replaceable> |
+  COMPRESSION <replaceable class="parameter">compression_access_method</replaceable> [ WITH (<replaceable class="parameter">compression_am_options</replaceable>) ] |
   REFERENCES <replaceable class="parameter">reftable</replaceable> [ ( <replaceable class="parameter">refcolumn</replaceable> ) ] [ MATCH FULL | MATCH PARTIAL | MATCH SIMPLE ]
     [ ON DELETE <replaceable class="parameter">referential_action</replaceable> ] [ ON UPDATE <replaceable class="parameter">referential_action</replaceable> ] }
 [ DEFERRABLE | NOT DEFERRABLE ] [ INITIALLY DEFERRED | INITIALLY IMMEDIATE ]
@@ -968,11 +968,13 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
    </varlistentry>
 
    <varlistentry>
-    <term><literal>COMPRESSION <replaceable class="parameter">compression_access_method</replaceable></literal></term>
+    <term><literal>COMPRESSION <replaceable class="parameter">compression_access_method</replaceable> [ WITH (<replaceable class="parameter">compression_am_options</replaceable>) ]</literal></term>
     <listitem>
      <para>
-      This clause adds compression to a column. Currently we can give the built-in
-      compression method.
+      This clause adds compression to a column. Compression method could be
+      created with <xref linkend="sql-create-access-method"/>. If compression
+      method has options they could be specified by <literal>WITH</literal>
+      parameter.
      </para>
     </listitem>
    </varlistentry>
diff --git a/doc/src/sgml/storage.sgml b/doc/src/sgml/storage.sgml
index ecc3359f64..b1bec9f1b2 100644
--- a/doc/src/sgml/storage.sgml
+++ b/doc/src/sgml/storage.sgml
@@ -396,7 +396,9 @@ Further details appear in <xref linkend="storage-toast-inmemory"/>.
 The default compression technique used for either in-line or out-of-line compressed
 data is a fairly simple and very fast member
 of the LZ family of compression techniques.  See
-<filename>src/common/pg_lzcompress.c</filename> for the details.
+<filename>src/common/pg_lzcompress.c</filename> for the details. Also custom
+compressions could be used. Look at <xref linkend="compression-am"/> for
+more information.
 </para>
 
 <sect2 id="storage-toast-ondisk">
diff --git a/src/backend/access/common/detoast.c b/src/backend/access/common/detoast.c
index 0a6592fbf9..aa77b1bddd 100644
--- a/src/backend/access/common/detoast.c
+++ b/src/backend/access/common/detoast.c
@@ -451,12 +451,12 @@ toast_decompress_datum(struct varlena *attr)
 
 	if (VARATT_IS_CUSTOM_COMPRESSED(attr))
 	{
-		CompressionAmOptions cmoptions;
+		CompressionAmOptions *cmoptions;
 		toast_compress_header_custom *hdr;
 
-		hdr = (toast_compress_header_custom *) attr;
-		lookup_compression_am_options(hdr->cmid, &cmoptions);
-		result = cmoptions.amroutine->cmdecompress(&cmoptions, attr);
+		hdr = (toast_compress_header_custom *)attr;
+		cmoptions = lookup_compression_am_options(hdr->cmid);
+		result = cmoptions->amroutine->cmdecompress(cmoptions, attr);
 	}
 	else
 	{
@@ -497,16 +497,15 @@ toast_decompress_datum_slice(struct varlena *attr, int32 slicelength)
 
 	if (VARATT_IS_CUSTOM_COMPRESSED(attr))
 	{
-		CompressionAmOptions cmoptions;
+		CompressionAmOptions *cmoptions;
 		toast_compress_header_custom *hdr;
 
-		hdr = (toast_compress_header_custom *) attr;
-		lookup_compression_am_options(hdr->cmid, &cmoptions);
-		if (cmoptions.amroutine->cmdecompress_slice)
-			result = cmoptions.amroutine->cmdecompress_slice(&cmoptions, attr,
-															 slicelength);
+		hdr = (toast_compress_header_custom *)attr;
+		cmoptions = lookup_compression_am_options(hdr->cmid);
+		if (cmoptions->amroutine->cmdecompress_slice)
+			result = cmoptions->amroutine->cmdecompress_slice(cmoptions, attr, slicelength);
 		else
-			result = cmoptions.amroutine->cmdecompress(&cmoptions, attr);
+			result = cmoptions->amroutine->cmdecompress(cmoptions, attr);
 	}
 	else
 	{
diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index 8ccc228a8c..faa5deb812 100644
--- a/src/backend/access/common/reloptions.c
+++ b/src/backend/access/common/reloptions.c
@@ -1336,11 +1336,99 @@ 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 ListCell *a, const ListCell *b)
+{
+	DefElem    *da = (DefElem *) lfirst(a);
+	DefElem    *db = (DefElem *) lfirst(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)
+		list_sort(options, compare_options);
+	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/common/toast_internals.c b/src/backend/access/common/toast_internals.c
index 38d273e8bd..62a09903b8 100644
--- a/src/backend/access/common/toast_internals.c
+++ b/src/backend/access/common/toast_internals.c
@@ -35,6 +35,9 @@
 #include "utils/snapmgr.h"
 #include "utils/syscache.h"
 
+HTAB *amoptions_cache = NULL;
+MemoryContext amoptions_cache_mcxt = NULL;
+
 static bool toastrel_valueid_exists(Relation toastrel, Oid valueid);
 static bool toastid_valueid_exists(Oid toastrelid, Oid valueid);
 
@@ -45,8 +48,7 @@ static bool toastid_valueid_exists(Oid toastrelid, Oid valueid);
  * for decompression.
  * --------
  */
-void
-toast_set_compressed_datum_info(struct varlena *val, Oid cmid, int32 rawsize)
+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);
@@ -66,21 +68,21 @@ toast_set_compressed_datum_info(struct varlena *val, Oid cmid, int32 rawsize)
  * ----------
  */
 Datum
-toast_compress_datum(Datum value, Oid amoid)
+toast_compress_datum(Datum value, Oid acoid)
 {
 	struct varlena *tmp = NULL;
 	int32 valsize;
-	CompressionAmOptions cmoptions;
+	CompressionAmOptions *cmoptions = NULL;
 
 	Assert(!VARATT_IS_EXTERNAL(DatumGetPointer(value)));
 	Assert(!VARATT_IS_COMPRESSED(DatumGetPointer(value)));
 
 	/* Fallback to default compression if not specified */
-	if (!OidIsValid(amoid))
-		amoid = DefaultCompressionOid;
+	if (!OidIsValid(acoid))
+		acoid = DefaultCompressionOid;
 
-	lookup_compression_am_options(amoid, &cmoptions);
-	tmp = cmoptions.amroutine->cmcompress(&cmoptions, (const struct varlena *) value);
+	cmoptions = lookup_compression_am_options(acoid);
+	tmp = cmoptions->amroutine->cmcompress(cmoptions, (const struct varlena *)value);
 	if (!tmp)
 		return PointerGetDatum(NULL);
 
@@ -100,7 +102,7 @@ toast_compress_datum(Datum value, Oid amoid)
 	if (VARSIZE(tmp) < valsize - 2)
 	{
 		/* successful compression */
-		toast_set_compressed_datum_info(tmp, amoid, valsize);
+		toast_set_compressed_datum_info(tmp, cmoptions->acoid, valsize);
 		return PointerGetDatum(tmp);
 	}
 
@@ -649,22 +651,157 @@ 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.
+ */
+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
  *
  * Get compression handler routine.
  */
-void
-lookup_compression_am_options(Oid acoid, CompressionAmOptions *result)
+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;
 
-	amhandler = get_am_handler_oid(acoid, AMTYPE_COMPRESSION, false);
-	result->amroutine = InvokeCompressionAmHandler(amhandler);
-	result->acstate = result->amroutine->cminitstate ?
-			result->amroutine->cminitstate(acoid) : NULL;
+	/*
+	 * 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
+ */
+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/compression/cm_pglz.c b/src/backend/access/compression/cm_pglz.c
index 7541b777ef..8203605966 100644
--- a/src/backend/access/compression/cm_pglz.c
+++ b/src/backend/access/compression/cm_pglz.c
@@ -18,17 +18,83 @@
 #include "nodes/parsenodes.h"
 #include "utils/builtins.h"
 
+#define PGLZ_OPTIONS_COUNT 6
+
+static char *PGLZ_options[PGLZ_OPTIONS_COUNT] = {
+	"min_input_size",
+	"max_input_size",
+	"min_comp_rate",
+	"first_success_by",
+	"match_size_good",
+	"match_size_drop"
+};
+
+/*
+ * Convert value from reloptions to int32, and report if it is not correct.
+ * Also checks parameter names
+ */
+static int32
+parse_option(char *name, char *value)
+{
+	int			i;
+
+	for (i = 0; i < PGLZ_OPTIONS_COUNT; i++)
+	{
+		if (strcmp(PGLZ_options[i], name) == 0)
+			return pg_atoi(value, 4, 0);
+	}
+
+	ereport(ERROR,
+			(errcode(ERRCODE_UNDEFINED_PARAMETER),
+			 errmsg("unexpected parameter for pglz: \"%s\"", name)));
+}
+
+/*
+ * Check PGLZ options if specified
+ */
+static void
+pglz_cmcheck(Form_pg_attribute att, List *options)
+{
+	ListCell   *lc;
+
+	foreach(lc, options)
+	{
+		DefElem    *def = (DefElem *) lfirst(lc);
+
+		parse_option(def->defname, defGetString(def));
+	}
+}
+
 /*
  * Configure PGLZ_Strategy struct for compression function
  */
 static void *
-pglz_cminitstate(Oid acoid)
+pglz_cminitstate(Oid acoid, List *options)
 {
+	ListCell   *lc;
 	PGLZ_Strategy *strategy = palloc(sizeof(PGLZ_Strategy));
 
 	/* initialize with default strategy values */
 	memcpy(strategy, PGLZ_strategy_default, sizeof(PGLZ_Strategy));
-
+	foreach(lc, options)
+	{
+		DefElem    *def = (DefElem *) lfirst(lc);
+		int32		val = parse_option(def->defname, defGetString(def));
+
+		/* fill the strategy */
+		if (strcmp(def->defname, "min_input_size") == 0)
+			strategy->min_input_size = val;
+		else if (strcmp(def->defname, "max_input_size") == 0)
+			strategy->max_input_size = val;
+		else if (strcmp(def->defname, "min_comp_rate") == 0)
+			strategy->min_comp_rate = val;
+		else if (strcmp(def->defname, "first_success_by") == 0)
+			strategy->first_success_by = val;
+		else if (strcmp(def->defname, "match_size_good") == 0)
+			strategy->match_size_good = val;
+		else if (strcmp(def->defname, "match_size_drop") == 0)
+			strategy->match_size_drop = val;
+	}
 	return (void *) strategy;
 }
 
@@ -114,6 +180,7 @@ pglzhandler(PG_FUNCTION_ARGS)
 {
 	CompressionAmRoutine *routine = makeNode(CompressionAmRoutine);
 
+	routine->cmcheck = pglz_cmcheck;
 	routine->cminitstate = pglz_cminitstate;
 	routine->cmcompress = pglz_cmcompress;
 	routine->cmdecompress = pglz_cmdecompress;
diff --git a/src/backend/access/compression/cm_zlib.c b/src/backend/access/compression/cm_zlib.c
index b1ba33e90b..e9bf2772e2 100644
--- a/src/backend/access/compression/cm_zlib.c
+++ b/src/backend/access/compression/cm_zlib.c
@@ -30,14 +30,100 @@ typedef struct
 	unsigned int	dictlen;
 } zlib_state;
 
+/*
+ * Check options if specified. All validation is located here so
+ * we don't need do it again in cminitstate function.
+ */
+static void
+zlib_cmcheck(Form_pg_attribute att, List *options)
+{
+	ListCell	*lc;
+	foreach(lc, options)
+	{
+		DefElem    *def = (DefElem *) lfirst(lc);
+
+		if (strcmp(def->defname, "level") == 0)
+		{
+			int8 level = pg_atoi(defGetString(def), sizeof(int8), 0);
+			if (level < 0 || level > 9)
+				ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("unexpected value for zlib compression level: \"%s\"",
+								defGetString(def)),
+					 errhint("expected value between 0 and 9")
+					));
+		}
+		else if (strcmp(def->defname, "dict") == 0)
+		{
+			int		ntokens = 0;
+			char   *val,
+				   *tok;
+
+			val = pstrdup(defGetString(def));
+			if (strlen(val) > (ZLIB_MAX_DICTIONARY_LENGTH - 1))
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						(errmsg("zlib dictionary length should be less than %d", ZLIB_MAX_DICTIONARY_LENGTH))));
+
+			while ((tok = strtok(val, ZLIB_DICTIONARY_DELIM)) != NULL)
+			{
+				ntokens++;
+				val = NULL;
+			}
+
+			if (ntokens < 2)
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						(errmsg("zlib dictionary is too small"))));
+		}
+		else
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_PARAMETER),
+					 errmsg("unexpected parameter for zlib: \"%s\"", def->defname)));
+	}
+}
+
 static void *
-zlib_cminitstate(Oid acoid)
+zlib_cminitstate(Oid acoid, List *options)
 {
 	zlib_state		*state = NULL;
 
 	state = palloc0(sizeof(zlib_state));
 	state->level = Z_DEFAULT_COMPRESSION;
 
+	if (list_length(options) > 0)
+	{
+		ListCell	*lc;
+
+		foreach(lc, options)
+		{
+			DefElem    *def = (DefElem *) lfirst(lc);
+
+			if (strcmp(def->defname, "level") == 0)
+				state->level = pg_atoi(defGetString(def), sizeof(int), 0);
+			else if (strcmp(def->defname, "dict") == 0)
+			{
+				char   *val,
+					   *tok;
+
+				val = pstrdup(defGetString(def));
+
+				/* Fill the zlib dictionary */
+				while ((tok = strtok(val, ZLIB_DICTIONARY_DELIM)) != NULL)
+				{
+					int len = strlen(tok);
+					memcpy((void *) (state->dict + state->dictlen), tok, len);
+					state->dictlen += len + 1;
+					Assert(state->dictlen <= ZLIB_MAX_DICTIONARY_LENGTH);
+
+					/* add space as dictionary delimiter */
+					state->dict[state->dictlen - 1] = ' ';
+					val = NULL;
+				}
+			}
+		}
+	}
+
 	return state;
 }
 
@@ -153,6 +239,7 @@ zlibhandler(PG_FUNCTION_ARGS)
 #else
 	CompressionAmRoutine *routine = makeNode(CompressionAmRoutine);
 
+	routine->cmcheck = zlib_cmcheck;
 	routine->cminitstate = zlib_cminitstate;
 	routine->cmcompress = zlib_cmcompress;
 	routine->cmdecompress = zlib_cmdecompress;
diff --git a/src/backend/access/compression/cmapi.c b/src/backend/access/compression/cmapi.c
index 3cdce17276..504da0638f 100644
--- a/src/backend/access/compression/cmapi.c
+++ b/src/backend/access/compression/cmapi.c
@@ -18,6 +18,7 @@
 #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"
 
@@ -41,3 +42,55 @@ InvokeCompressionAmHandler(Oid 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 9d52fa1131..38ddc9f69e 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -72,7 +72,7 @@
 
 
 static HeapTuple heap_prepare_insert(Relation relation, HeapTuple tup,
-									 TransactionId xid, CommandId cid, int options);
+		TransactionId xid, CommandId cid, int options, BulkInsertState bistate);
 static XLogRecPtr log_heap_update(Relation reln, Buffer oldbuf,
 								  Buffer newbuf, HeapTuple oldtup,
 								  HeapTuple newtup, HeapTuple old_key_tuple,
@@ -1803,13 +1803,14 @@ UpdateXmaxHintBits(HeapTupleHeader tuple, Buffer buffer, TransactionId xid)
  * GetBulkInsertState - prepare status object for a bulk insert
  */
 BulkInsertState
-GetBulkInsertState(void)
+GetBulkInsertState(HTAB *preserved_am_info)
 {
 	BulkInsertState bistate;
 
 	bistate = (BulkInsertState) palloc(sizeof(BulkInsertStateData));
 	bistate->strategy = GetAccessStrategy(BAS_BULKWRITE);
 	bistate->current_buf = InvalidBuffer;
+	bistate->preserved_am_info = preserved_am_info;
 	return bistate;
 }
 
@@ -1871,7 +1872,7 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
 	 * Note: below this point, heaptup is the data we actually intend to store
 	 * into the relation; tup is the caller's original untoasted data.
 	 */
-	heaptup = heap_prepare_insert(relation, tup, xid, cid, options);
+	heaptup = heap_prepare_insert(relation, tup, xid, cid, options, bistate);
 
 	/*
 	 * Find buffer to insert this tuple into.  If the page is all visible,
@@ -2041,7 +2042,7 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
  */
 static HeapTuple
 heap_prepare_insert(Relation relation, HeapTuple tup, TransactionId xid,
-					CommandId cid, int options)
+					CommandId cid, int options, BulkInsertState bistate)
 {
 	/*
 	 * Parallel operations are required to be strictly read-only in a parallel
@@ -2082,7 +2083,8 @@ heap_prepare_insert(Relation relation, HeapTuple tup, TransactionId xid,
 			 || RelationGetDescr(relation)->tdflags & TD_ATTR_CUSTOM_COMPRESSED
 			 || HeapTupleHasCustomCompressed(tup)
 			 || tup->t_len > TOAST_TUPLE_THRESHOLD)
-		return heap_toast_insert_or_update(relation, tup, NULL, options);
+		return heap_toast_insert_or_update(relation, tup, NULL, options,
+										   bistate ? bistate->preserved_am_info : NULL);
 	else
 		return tup;
 }
@@ -2130,7 +2132,7 @@ heap_multi_insert(Relation relation, TupleTableSlot **slots, int ntuples,
 		slots[i]->tts_tableOid = RelationGetRelid(relation);
 		tuple->t_tableOid = slots[i]->tts_tableOid;
 		heaptuples[i] = heap_prepare_insert(relation, tuple, xid, cid,
-											options);
+											options, bistate);
 	}
 
 	/*
@@ -3511,7 +3513,7 @@ l2:
 		if (need_toast)
 		{
 			/* Note we always use WAL and FSM during updates */
-			heaptup = heap_toast_insert_or_update(relation, newtup, &oldtup, 0);
+			heaptup = heap_toast_insert_or_update(relation, newtup, &oldtup, 0, NULL);
 			newtupsize = MAXALIGN(heaptup->t_len);
 		}
 		else
diff --git a/src/backend/access/heap/heaptoast.c b/src/backend/access/heap/heaptoast.c
index e7fa3f6683..2ccc9893ae 100644
--- a/src/backend/access/heap/heaptoast.c
+++ b/src/backend/access/heap/heaptoast.c
@@ -94,7 +94,7 @@ heap_toast_delete(Relation rel, HeapTuple oldtup, bool is_speculative)
  */
 HeapTuple
 heap_toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
-							int options)
+							int options, HTAB *preserved_am_info)
 {
 	HeapTuple	result_tuple;
 	TupleDesc	tupleDesc;
@@ -154,7 +154,7 @@ heap_toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
 		ttc.ttc_oldisnull = toast_oldisnull;
 	}
 	ttc.ttc_attr = toast_attr;
-	toast_tuple_init(&ttc);
+	toast_tuple_init(&ttc, preserved_am_info);
 
 	/* ----------
 	 * Compress and/or save external until data fits into target length
@@ -531,7 +531,7 @@ toast_flatten_tuple_to_datum(HeapTupleHeader tup,
 	heap_fill_tuple(tupleDesc,
 					toast_values,
 					toast_isnull,
-					(char *) new_data + new_header_len,
+					(char *)new_data + new_header_len,
 					new_data_len,
 					&(new_data->t_infomask),
 					&(new_data->t_infomask2),
diff --git a/src/backend/access/heap/rewriteheap.c b/src/backend/access/heap/rewriteheap.c
index 39e33763df..ede9fda72f 100644
--- a/src/backend/access/heap/rewriteheap.c
+++ b/src/backend/access/heap/rewriteheap.c
@@ -646,7 +646,7 @@ raw_heap_insert(RewriteState state, HeapTuple tup)
 		options |= HEAP_INSERT_NO_LOGICAL;
 
 		heaptup = heap_toast_insert_or_update(state->rs_new_rel, tup, NULL,
-											  options);
+											  options, NULL);
 	}
 	else
 		heaptup = tup;
diff --git a/src/backend/access/table/toast_helper.c b/src/backend/access/table/toast_helper.c
index b33c925a4b..a882a145fb 100644
--- a/src/backend/access/table/toast_helper.c
+++ b/src/backend/access/table/toast_helper.c
@@ -38,7 +38,7 @@
  * perform any initialization of the array before calling this function.
  */
 void
-toast_tuple_init(ToastTupleContext *ttc)
+toast_tuple_init(ToastTupleContext *ttc, HTAB *preserved_am_info)
 {
 	TupleDesc	tupleDesc = ttc->ttc_rel->rd_att;
 	int			numAttrs = tupleDesc->natts;
@@ -120,12 +120,30 @@ toast_tuple_init(ToastTupleContext *ttc)
 		 */
 		if (att->attlen == -1)
 		{
+			List *preserved_amoids = NIL;
+
 			/*
 			 * If the table's attribute says PLAIN always, force it so.
 			 */
 			if (att->attstorage == TYPSTORAGE_PLAIN)
 				ttc->ttc_attr[i].tai_colflags |= TOASTCOL_IGNORE;
 
+			/*
+			 * If it's ALTER SET COMPRESSION command we should check that
+			 * access method of compression in preserved list.
+			 */
+			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;
+			}
+
 			/*
 			 * 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
@@ -145,6 +163,64 @@ toast_tuple_init(ToastTupleContext *ttc)
 				ttc->ttc_attr[i].tai_colflags |= TOASTCOL_NEEDS_FREE;
 				ttc->ttc_flags |= (TOAST_NEEDS_CHANGE | TOAST_NEEDS_FREE);
 			}
+			/*
+			 * 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.
+			 */
+			else if (VARATT_IS_CUSTOM_COMPRESSED(new_value))
+			{
+				bool storage_ok;
+				toast_compress_header_custom *hdr;
+
+				storage_ok = (att->attstorage == TYPSTORAGE_MAIN ||
+							  att->attstorage == TYPSTORAGE_PLAIN);
+				hdr = (toast_compress_header_custom *) new_value;
+
+				if (!storage_ok || hdr->cmid != att->attcompression)
+				{
+					if (storage_ok &&
+						OidIsValid(att->attcompression) &&
+						attr_compression_options_are_equal(att->attcompression,
+														   hdr->cmid))
+					{
+						struct varlena *tmpval = NULL;
+
+						/* identical compression, just change Oid to new one */
+						tmpval = palloc(VARSIZE(new_value));
+						memcpy(tmpval, new_value, VARSIZE(new_value));
+						TOAST_COMPRESS_SET_CMID(tmpval, att->attcompression);
+						new_value = tmpval;
+						ttc->ttc_attr[i].tai_colflags |= TOASTCOL_NEEDS_FREE;
+						ttc->ttc_flags |= (TOAST_NEEDS_CHANGE | TOAST_NEEDS_FREE);
+					}
+					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))
+						{
+							new_value = detoast_attr(new_value);
+							ttc->ttc_attr[i].tai_colflags |= TOASTCOL_NEEDS_FREE;
+							ttc->ttc_flags |= (TOAST_NEEDS_CHANGE | TOAST_NEEDS_FREE);
+						}
+					}
+					else
+					{
+						/* just decompress the value */
+						new_value = detoast_attr(new_value);
+						ttc->ttc_attr[i].tai_colflags |= TOASTCOL_NEEDS_FREE;
+						ttc->ttc_flags |= (TOAST_NEEDS_CHANGE | TOAST_NEEDS_FREE);
+					}
+
+					/* FIXME: Check the ttc_flags */
+					ttc->ttc_values[i] = PointerGetDatum(new_value);
+				}
+			}
 
 			/*
 			 * Remember the size of this attribute
diff --git a/src/backend/catalog/Makefile b/src/backend/catalog/Makefile
index 93cf6d4368..7b3a0a0efb 100644
--- a/src/backend/catalog/Makefile
+++ b/src/backend/catalog/Makefile
@@ -56,7 +56,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_data.h \
+	pg_statistic_ext.h pg_statistic_ext_data.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 \
diff --git a/src/backend/catalog/dependency.c b/src/backend/catalog/dependency.c
index b67d3e5666..99bac7839e 100644
--- a/src/backend/catalog/dependency.c
+++ b/src/backend/catalog/dependency.c
@@ -25,6 +25,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"
@@ -181,6 +182,7 @@ static const Oid object_classes[] = {
 	PublicationRelRelationId,	/* OCLASS_PUBLICATION_REL */
 	SubscriptionRelationId,		/* OCLASS_SUBSCRIPTION */
 	TransformRelationId,		/* OCLASS_TRANSFORM */
+	AttrCompressionRelationId	/* OCLASS_ATTR_COMPRESSION */
 };
 
 
@@ -1499,6 +1501,10 @@ doDeletion(const ObjectAddress *object, int flags)
 			DropObjectById(object);
 			break;
 
+		case OCLASS_ATTR_COMPRESSION:
+			RemoveAttributeCompression(object->objectId);
+			break;
+
 			/*
 			 * These global object types are not supported here.
 			 */
@@ -2843,6 +2849,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/objectaddress.c b/src/backend/catalog/objectaddress.c
index 6dfe1be2cc..b9e22beb92 100644
--- a/src/backend/catalog/objectaddress.c
+++ b/src/backend/catalog/objectaddress.c
@@ -26,6 +26,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"
@@ -616,6 +617,19 @@ static const ObjectPropertyType ObjectProperty[] =
 		OBJECT_USER_MAPPING,
 		false
 	},
+	{
+		"compression",
+		AttrCompressionRelationId,
+		AttrCompressionIndexId,
+		ATTCOMPRESSIONOID,
+		-1,
+		InvalidAttrNumber,
+		InvalidAttrNumber,
+		InvalidAttrNumber,
+		InvalidAttrNumber,
+		-1,
+		false
+	}
 };
 
 /*
@@ -3907,6 +3921,37 @@ getObjectDescription(const ObjectAddress *object, bool missing_ok)
 				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, false);
+
+					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;
@@ -4479,6 +4524,10 @@ getObjectTypeDescription(const ObjectAddress *object, bool missing_ok)
 			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.
@@ -5155,6 +5204,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/alter.c b/src/backend/commands/alter.c
index b11ebf0f61..5c30758197 100644
--- a/src/backend/commands/alter.c
+++ b/src/backend/commands/alter.c
@@ -663,6 +663,7 @@ AlterObjectNamespace_oid(Oid classId, Oid objid, Oid nspOid,
 		case OCLASS_PUBLICATION_REL:
 		case OCLASS_SUBSCRIPTION:
 		case OCLASS_TRANSFORM:
+		case OCLASS_ATTR_COMPRESSION:
 			/* ignore object types that don't have schema-qualified names */
 			break;
 
diff --git a/src/backend/commands/compressioncmds.c b/src/backend/commands/compressioncmds.c
index 3d56f7399f..9cf57ae0cd 100644
--- a/src/backend/commands/compressioncmds.c
+++ b/src/backend/commands/compressioncmds.c
@@ -22,7 +22,7 @@
 #include "catalog/catalog.h"
 #include "catalog/dependency.h"
 #include "catalog/indexing.h"
-#include "catalog/pg_am.h"
+#include "catalog/pg_attr_compression_d.h"
 #include "catalog/pg_am_d.h"
 #include "catalog/pg_collation_d.h"
 #include "catalog/pg_depend.h"
@@ -37,49 +37,636 @@
 #include "utils/syscache.h"
 #include "utils/snapmgr.h"
 
+/* Set by pg_upgrade_support functions */
+Oid			binary_upgrade_next_attr_compression_oid = InvalidOid;
+
 /*
- * Link compression with an attribute.
+ * 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 = table_open(DependRelationId, lock);
+
+	ScanKeyInit(&key[0],
+				Anum_pg_depend_classid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(RelationRelationId));
+	ScanKeyInit(&key[1],
+				Anum_pg_depend_objid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(attrelid));
+	ScanKeyInit(&key[2],
+				Anum_pg_depend_objsubid,
+				BTEqualStrategyNumber, F_INT4EQ,
+				Int32GetDatum((int32) attnum));
+
+	scan = systable_beginscan(rel, DependDependerIndexId, true,
+							  NULL, 3, key);
+
+	while (HeapTupleIsValid(tup = systable_getnext(scan)))
+	{
+		Form_pg_depend depform = (Form_pg_depend) GETSTRUCT(tup);
+
+		if (depform->refclassid == AttrCompressionRelationId)
+		{
+			Assert(IsBuiltinCompression(depform->refobjid));
+			amoid = GetAttrCompressionAmOid(depform->refobjid);
+			*amoids = list_append_unique_oid(*amoids, amoid);
+		}
+	}
+
+	systable_endscan(scan);
+	table_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 = table_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];
+		char	   *amname;
+
+		heap_deform_tuple(tuple, RelationGetDescr(rel), values, nulls);
+		acoid = DatumGetObjectId(values[Anum_pg_attr_compression_acoid - 1]);
+		amname = NameStr(*DatumGetName(values[Anum_pg_attr_compression_acname - 1]));
+		tup_amoid = get_am_oid(amname, 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 if (DatumGetPointer(acoptions) != NULL)
+		{
+			bool		equal;
+
+			/* check if arrays for WITH options are equal */
+			equal = DatumGetBool(CallerFInfoFunctionCall2(
+						array_eq, &arrayeq_info, DEFAULT_COLLATION_OID, acoptions,
+						values[Anum_pg_attr_compression_acoptions - 1]));
+
+			if (equal)
+				result = acoid;
+		}
+
+		if (previous_amoids == NULL && OidIsValid(result))
+			break;
+	}
+
+	systable_endscan(scan);
+	table_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
-GetAttributeCompression(Form_pg_attribute att, char *compression)
+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 == TYPSTORAGE_PLAIN)
+	if (att->attstorage == 'p')
 		return InvalidOid;
 
 	/* Fallback to default compression if it's not specified */
 	if (compression == NULL)
 		return DefaultCompressionOid;
 
-	return get_compression_am_oid(compression, false);
+	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);
+
+	/* no rewrite by default */
+	if (need_rewrite != NULL)
+		*need_rewrite = false;
+
+	if (IsBinaryUpgrade)
+	{
+		/* Skip the rewrite checks and searching of identical compression */
+		goto add_tuple;
+	}
+
+	/*
+	 * 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)
+		{
+			Assert(preserved_amoids != NULL);
+
+			if (compression->preserve == NIL)
+				*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 (list_length(previous_amoids) != 0)
+					*need_rewrite = true;
+			}
+		}
+
+		/* Cleanup */
+		list_free(previous_amoids);
+	}
+
+	/* Return Oid if we already found identical compression on this column */
+	if (OidIsValid(acoid))
+	{
+		if (DatumGetPointer(arropt) != NULL)
+			pfree(DatumGetPointer(arropt));
+
+		return acoid;
+	}
+
+add_tuple:
+	/* 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 = table_open(AttrCompressionRelationId, RowExclusiveLock);
+
+	if (IsBinaryUpgrade)
+	{
+		/* acoid should be found in some cases */
+		if (binary_upgrade_next_attr_compression_oid < FirstNormalObjectId &&
+			(!OidIsValid(acoid) || binary_upgrade_next_attr_compression_oid != acoid))
+			elog(ERROR, "could not link to built-in attribute compression");
+
+		acoid = binary_upgrade_next_attr_compression_oid;
+	}
+	else
+	{
+		acoid = GetNewOidWithIndex(rel, AttrCompressionIndexId,
+									Anum_pg_attr_compression_acoid);
+
+	}
+
+	if (acoid < FirstNormalObjectId)
+	{
+		/* this is built-in attribute compression */
+		table_close(rel, RowExclusiveLock);
+		return acoid;
+	}
+
+	/* 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);
+	table_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;
 }
 
 /*
- * Get compression name by attribute compression Oid.
+ * Remove the attribute compression record from pg_attr_compression.
  */
-char *
-GetCompressionName(Oid amoid)
+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 = table_open(AttrCompressionRelationId, RowExclusiveLock);
+	CatalogTupleDelete(relation, &tup->t_self);
+	table_close(relation, RowExclusiveLock);
+	ReleaseSysCache(tup);
+}
+
+/*
+ * CleanupAttributeCompression
+ *
+ * Remove entries in pg_attr_compression of the column 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);
+	Assert(!IsBinaryUpgrade);
+
+	rel = table_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);
+	table_close(rel, RowExclusiveLock);
+
+	/*
+	 * Now remove dependencies between attribute compression (dependent)
+	 * and column.
+	 */
+	rel = table_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);
+	}
+	table_close(rel, RowExclusiveLock);
+
+	/* Now remove dependencies with builtin compressions */
+	rel = table_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);
+	table_close(rel, RowExclusiveLock);
+}
+
+/*
+ * Construct ColumnCompression node by attribute compression Oid.
+ */
+ColumnCompression *
+MakeColumnCompression(Oid acoid)
 {
 	HeapTuple	tuple;
-	Form_pg_am amform;
-	char	  *amname;
+	Form_pg_attr_compression acform;
+	ColumnCompression *node;
 
-	if (!OidIsValid(amoid))
+	if (!OidIsValid(acoid))
 		return NULL;
 
-	/* Get handler function OID for the access method */
-	tuple = SearchSysCache1(AMOID, ObjectIdGetDatum(amoid));
+	tuple = SearchSysCache1(ATTCOMPRESSIONOID, ObjectIdGetDatum(acoid));
 	if (!HeapTupleIsValid(tuple))
-		elog(ERROR, "cache lookup failed for attribute compression %u", amoid);
-
-	amform = (Form_pg_am) GETSTRUCT(tuple);
-	amname = pstrdup(NameStr(amform->amname));
+		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);
 
-	return amname;
+	/*
+	 * 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 = table_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);
+	table_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/copy.c b/src/backend/commands/copy.c
index db7d24a511..c1fcadebe2 100644
--- a/src/backend/commands/copy.c
+++ b/src/backend/commands/copy.c
@@ -2394,7 +2394,7 @@ CopyMultiInsertBufferInit(ResultRelInfo *rri)
 	buffer = (CopyMultiInsertBuffer *) palloc(sizeof(CopyMultiInsertBuffer));
 	memset(buffer->slots, 0, sizeof(TupleTableSlot *) * MAX_BUFFERED_TUPLES);
 	buffer->resultRelInfo = rri;
-	buffer->bistate = GetBulkInsertState();
+	buffer->bistate = GetBulkInsertState(NULL);
 	buffer->nused = 0;
 
 	return buffer;
@@ -2974,7 +2974,7 @@ CopyFrom(CopyState cstate)
 	{
 		singleslot = table_slot_create(resultRelInfo->ri_RelationDesc,
 									   &estate->es_tupleTable);
-		bistate = GetBulkInsertState();
+		bistate = GetBulkInsertState(NULL);
 	}
 
 	has_before_insert_row_trig = (resultRelInfo->ri_TrigDesc &&
diff --git a/src/backend/commands/createas.c b/src/backend/commands/createas.c
index d53ec952d0..c593f236a5 100644
--- a/src/backend/commands/createas.c
+++ b/src/backend/commands/createas.c
@@ -552,7 +552,7 @@ intorel_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
 	myState->reladdr = intoRelationAddr;
 	myState->output_cid = GetCurrentCommandId(true);
 	myState->ti_options = TABLE_INSERT_SKIP_FSM;
-	myState->bistate = GetBulkInsertState();
+	myState->bistate = GetBulkInsertState(NULL);
 
 	/*
 	 * Valid smgr_targblock implies something already wrote to the relation.
diff --git a/src/backend/commands/event_trigger.c b/src/backend/commands/event_trigger.c
index 7844880170..8d6e837274 100644
--- a/src/backend/commands/event_trigger.c
+++ b/src/backend/commands/event_trigger.c
@@ -1053,6 +1053,7 @@ EventTriggerSupportsObjectClass(ObjectClass objclass)
 		case OCLASS_PUBLICATION_REL:
 		case OCLASS_SUBSCRIPTION:
 		case OCLASS_TRANSFORM:
+		case OCLASS_ATTR_COMPRESSION:
 			return true;
 
 			/*
diff --git a/src/backend/commands/foreigncmds.c b/src/backend/commands/foreigncmds.c
index de31ddd1f3..78935294b6 100644
--- a/src/backend/commands/foreigncmds.c
+++ b/src/backend/commands/foreigncmds.c
@@ -49,50 +49,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
@@ -176,7 +132,7 @@ transformGenericOptions(Oid catalogId,
 		}
 	}
 
-	result = optionListToArray(resultOptions);
+	result = optionListToArray(resultOptions, false);
 
 	if (OidIsValid(fdwvalidator))
 	{
diff --git a/src/backend/commands/matview.c b/src/backend/commands/matview.c
index f80a9e96a9..37115e546b 100644
--- a/src/backend/commands/matview.c
+++ b/src/backend/commands/matview.c
@@ -458,7 +458,7 @@ transientrel_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
 	myState->transientrel = transientrel;
 	myState->output_cid = GetCurrentCommandId(true);
 	myState->ti_options = TABLE_INSERT_SKIP_FSM | TABLE_INSERT_FROZEN;
-	myState->bistate = GetBulkInsertState();
+	myState->bistate = GetBulkInsertState(NULL);
 
 	/*
 	 * Valid smgr_targblock implies something already wrote to the relation.
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index d71bfdb7c6..8ef0434ebb 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -35,6 +35,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"
@@ -168,6 +169,8 @@ typedef struct AlteredTableInfo
 	List	   *afterStmts;		/* List of utility command parsetrees */
 	bool		verify_new_notnull; /* T if we should recheck NOT NULL */
 	bool		new_notnull;	/* T if we added new NOT NULL constraints */
+	HTAB	   *preservedAmInfo;	/* Hash table for preserved compression
+									 * methods */
 	int			rewrite;		/* Reason for forced rewrite, if any */
 	Oid			newTableSpace;	/* new tablespace; 0 means no change */
 	bool		chgPersistence; /* T if SET LOGGED/UNLOGGED is used */
@@ -394,6 +397,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 ATPrepDropNotNull(Relation rel, bool recurse, bool recursing);
 static ObjectAddress ATExecDropNotNull(Relation rel, const char *colName, LOCKMODE lockmode);
 static void ATPrepSetNotNull(List **wqueue, Relation rel,
@@ -531,6 +536,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 index_copy_data(Relation rel, RelFileNode newrnode);
 static const char *storage_name(char c);
@@ -585,6 +592,7 @@ ObjectAddress
 DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 			   ObjectAddress *typaddress, const char *queryString)
 {
+	int			i;
 	char		relname[NAMEDATALEN];
 	Oid			namespaceId;
 	Oid			relationId;
@@ -859,8 +867,8 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 
 		if (!IsBinaryUpgrade &&
 			(relkind == RELKIND_RELATION || relkind == RELKIND_PARTITIONED_TABLE))
-			attr->attcompression = GetAttributeCompression(attr,
-														colDef->compression);
+			attr->attcompression = CreateAttributeCompression(attr,
+										colDef->compression, NULL, NULL);
 		else
 			attr->attcompression = InvalidOid;
 	}
@@ -930,6 +938,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);
+		}
+	}
+
 	/*
 	 * Now add any newly specified column default and generation expressions
 	 * to the new relation.  These are passed to us in the form of raw
@@ -2386,16 +2411,15 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 
 				if (OidIsValid(attribute->attcompression))
 				{
-					char *compression = GetCompressionName(attribute->attcompression);
+					ColumnCompression *compression =
+					MakeColumnCompression(attribute->attcompression);
 
 					if (!def->compression)
 						def->compression = compression;
-					else if (strcmp(def->compression, compression))
-							ereport(ERROR,
-								(errcode(ERRCODE_DATATYPE_MISMATCH),
-									errmsg("column \"%s\" has a compression method conflict",
-										attributeName),
-									errdetail("%s versus %s", def->compression, compression)));
+					else
+						CheckCompressionMismatch(def->compression,
+												 compression,
+												 attributeName);
 				}
 
 				def->inhcount++;
@@ -2432,7 +2456,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 				def->collOid = attribute->attcollation;
 				def->constraints = NIL;
 				def->location = -1;
-				def->compression = GetCompressionName(attribute->attcompression);
+				def->compression = MakeColumnCompression(attribute->attcompression);
 				inhSchema = lappend(inhSchema, def);
 				newattmap->attnums[parent_attno - 1] = ++child_attno;
 			}
@@ -2645,14 +2669,9 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 				if (!def->compression)
 					def->compression = newdef->compression;
 				else if (newdef->compression)
-				{
-					if (strcmp(def->compression, newdef->compression))
-						ereport(ERROR,
-							(errcode(ERRCODE_DATATYPE_MISMATCH),
-								errmsg("column \"%s\" has a compression method conflict",
-									attributeName),
-								errdetail("%s versus %s", def->compression, newdef->compression)));
-				}
+					CheckCompressionMismatch(def->compression,
+											 newdef->compression,
+											 attributeName);
 
 				/* Mark the column as locally defined */
 				def->is_local = true;
@@ -3799,6 +3818,7 @@ AlterTableGetLockLevel(List *cmds)
 				 */
 			case AT_GenericOptions:
 			case AT_AlterColumnGenericOptions:
+			case AT_SetCompression:
 				cmd_lockmode = AccessExclusiveLock;
 				break;
 
@@ -4306,6 +4326,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 */
@@ -4699,6 +4720,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);
@@ -5141,7 +5167,7 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode)
 	if (newrel)
 	{
 		mycid = GetCurrentCommandId(true);
-		bistate = GetBulkInsertState();
+		bistate = GetBulkInsertState(tab->preservedAmInfo);
 		ti_options = TABLE_INSERT_SKIP_FSM;
 	}
 	else
@@ -5463,6 +5489,24 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode)
 
 	FreeExecutorState(estate);
 
+	/* Remove old compression options */
+	if (tab->rewrite & AT_REWRITE_ALTER_COMPRESSION)
+	{
+		AttrCmPreservedInfo *pinfo;
+		HASH_SEQ_STATUS status;
+
+		Assert(tab->preservedAmInfo);
+		hash_seq_init(&status, tab->preservedAmInfo);
+		while ((pinfo = (AttrCmPreservedInfo *) hash_seq_search(&status)) != NULL)
+		{
+			CleanupAttributeCompression(tab->relid, pinfo->attnum,
+										pinfo->preserved_amoids);
+			list_free(pinfo->preserved_amoids);
+		}
+
+		hash_destroy(tab->preservedAmInfo);
+	}
+
 	table_close(oldrel, NoLock);
 	if (newrel)
 	{
@@ -6176,12 +6220,13 @@ ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel,
 	/* create attribute compresssion record */
 	if (rel->rd_rel->relkind == RELKIND_RELATION ||
 		rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
-		attribute.attcompression = GetAttributeCompression(&attribute,
-															  colDef->compression);
+		attribute.attcompression = CreateAttributeCompression(&attribute,
+															  colDef->compression,
+															  NULL, NULL);
 	else
 		attribute.attcompression = InvalidOid;
 
-	/* attribute.attacl is handled by InsertPgAttributeTuples() */
+	/* attribute.attacl is handled by InsertPgAttributeTuple() */
 
 	ReleaseSysCache(typeTuple);
 
@@ -6352,6 +6397,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
@@ -6477,6 +6523,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 = table_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);
+	table_close(rel, RowExclusiveLock);
+}
+
 /*
  * Install a column's dependency on its collation.
  */
@@ -6499,6 +6569,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 = table_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);
+	}
+	table_close(depRel, AccessShareLock);
+}
+
 /*
  * ALTER TABLE ALTER COLUMN DROP NOT NULL
  */
@@ -10591,6 +10761,45 @@ createForeignKeyCheckTriggers(Oid myRelOid, Oid refRelOid,
 						 indexOid, false);
 }
 
+/*
+ * Initialize hash table used to keep rewrite rules for
+ * compression changes in ALTER commands.
+ */
+static void
+setupCompressionRewriteRules(AlteredTableInfo *tab, const char *column,
+							 AttrNumber attnum, List *preserved_amoids)
+{
+	bool		found;
+	AttrCmPreservedInfo *pinfo;
+
+	Assert(!IsBinaryUpgrade);
+	tab->rewrite |= AT_REWRITE_ALTER_COMPRESSION;
+
+	/* initialize hash for oids */
+	if (tab->preservedAmInfo == NULL)
+	{
+		HASHCTL		ctl;
+
+		MemSet(&ctl, 0, sizeof(ctl));
+		ctl.keysize = sizeof(AttrNumber);
+		ctl.entrysize = sizeof(AttrCmPreservedInfo);
+		tab->preservedAmInfo =
+			hash_create("preserved access methods cache", 10, &ctl,
+						HASH_ELEM | HASH_BLOBS);
+	}
+	pinfo = (AttrCmPreservedInfo *) hash_search(tab->preservedAmInfo,
+												&attnum, HASH_ENTER, &found);
+
+	if (found)
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("cannot alter compression of column \"%s\" twice", column),
+				 errhint("Remove one of statements from the command.")));
+
+	pinfo->attnum = attnum;
+	pinfo->preserved_amoids = preserved_amoids;
+}
+
 /*
  * ALTER TABLE DROP CONSTRAINT
  *
@@ -11420,6 +11629,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:
 
 				/*
@@ -11530,7 +11745,8 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
 			  foundDep->refobjid == attTup->attcollation) &&
 			!(foundDep->refclassid == RelationRelationId &&
 			  foundDep->refobjid == RelationGetRelid(rel) &&
-			  foundDep->refobjsubid != 0)
+			  foundDep->refobjsubid != 0) &&
+			 foundDep->refclassid != AttrCompressionRelationId
 			)
 			elog(ERROR, "found unexpected dependency for column: %s",
 				 getObjectDescription(&foundObject, false));
@@ -11624,11 +11840,27 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
 	if (rel->rd_rel->relkind == RELKIND_RELATION ||
 		rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
 	{
-		/* Setup attribute compression */
-		if (attTup->attstorage == TYPSTORAGE_PLAIN)
+		/* Setup attribute compression for new attribute */
+		if (OidIsValid(attTup->attcompression) && attTup->attstorage == 'p')
+		{
+			/* Set up rewrite of table */
 			attTup->attcompression = InvalidOid;
-		else if (!OidIsValid(attTup->attcompression))
+			setupCompressionRewriteRules(tab, colName, attOldTup->attnum, NIL);
+		}
+		else if (OidIsValid(attTup->attcompression))
+		{
+			/* create new record */
+			ColumnCompression *compression = MakeColumnCompression(attTup->attcompression);
+
+			if (!IsBuiltinCompression(attTup->attcompression))
+				setupCompressionRewriteRules(tab, colName, attOldTup->attnum, NIL);
+
+			attTup->attcompression = CreateAttributeCompression(attTup, compression, NULL, NULL);
+		}
+		else if (attTup->attstorage != 'p')
 			attTup->attcompression = DefaultCompressionOid;
+		else
+			attTup->attcompression = InvalidOid;
 	}
 	else
 		attTup->attcompression = InvalidOid;
@@ -11641,6 +11873,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
 	 */
@@ -14810,6 +15047,86 @@ 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 = table_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)
+		setupCompressionRewriteRules(tab, column, atttableform->attnum,
+									 preserved_amoids);
+
+	atttableform->attcompression = acoid;
+	CatalogTupleUpdate(attrel, &atttuple->t_self, atttuple);
+
+	InvokeObjectPostAlterHook(RelationRelationId,
+							  RelationGetRelid(rel),
+							  atttableform->attnum);
+
+	ReleaseSysCache(atttuple);
+	table_close(attrel, RowExclusiveLock);
+
+	/* make changes visible */
+	CommandCounterIncrement();
+
+	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 7d71f4fb99..9dcd088e37 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -2930,7 +2930,7 @@ _copyColumnDef(const ColumnDef *from)
 
 	COPY_STRING_FIELD(colname);
 	COPY_NODE_FIELD(typeName);
-	COPY_STRING_FIELD(compression);
+	COPY_NODE_FIELD(compression);
 	COPY_SCALAR_FIELD(inhcount);
 	COPY_SCALAR_FIELD(is_local);
 	COPY_SCALAR_FIELD(is_not_null);
@@ -2950,6 +2950,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)
 {
@@ -5632,6 +5644,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 583a7414f0..880ff9cdaf 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -2589,7 +2589,7 @@ _equalColumnDef(const ColumnDef *a, const ColumnDef *b)
 {
 	COMPARE_STRING_FIELD(colname);
 	COMPARE_NODE_FIELD(typeName);
-	COMPARE_STRING_FIELD(compression);
+	COMPARE_NODE_FIELD(compression);
 	COMPARE_SCALAR_FIELD(inhcount);
 	COMPARE_SCALAR_FIELD(is_local);
 	COMPARE_SCALAR_FIELD(is_not_null);
@@ -2609,6 +2609,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)
 {
@@ -3684,6 +3694,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 92452c5fd6..b347341160 100644
--- a/src/backend/nodes/nodeFuncs.c
+++ b/src/backend/nodes/nodeFuncs.c
@@ -3817,6 +3817,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 a880a95ac6..2e4ac1375c 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -2834,7 +2834,7 @@ _outColumnDef(StringInfo str, const ColumnDef *node)
 
 	WRITE_STRING_FIELD(colname);
 	WRITE_NODE_FIELD(typeName);
-	WRITE_STRING_FIELD(compression);
+	WRITE_NODE_FIELD(compression);
 	WRITE_INT_FIELD(inhcount);
 	WRITE_BOOL_FIELD(is_local);
 	WRITE_BOOL_FIELD(is_not_null);
@@ -2852,6 +2852,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)
 {
@@ -4200,6 +4210,9 @@ outNode(StringInfo str, const void *obj)
 			case T_ColumnDef:
 				_outColumnDef(str, obj);
 				break;
+			case T_ColumnCompression:
+				_outColumnCompression(str, obj);
+				break;
 			case T_TypeName:
 				_outTypeName(str, obj);
 				break;
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 39f7be1854..54d311db49 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -415,6 +415,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 				transform_element_list transform_type_list
 				TriggerTransitions TriggerReferencing
 				publication_name_list
+				optCompressionParameters
 				vacuum_relation_list opt_vacuum_relation_list
 				drop_option_list
 
@@ -599,7 +600,9 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 %type <list>		hash_partbound
 %type <defelt>		hash_partbound_elem
 
-%type <str>	optColumnCompression
+%type <node>	optColumnCompression alterColumnCompression
+%type <str>		compressionClause
+%type <list>	optCompressionPreserve
 
 /*
  * Non-keyword token types.  These are hard-wired into the "flex" lexer.
@@ -2269,6 +2272,15 @@ alter_table_cmd:
 					n->missing_ok = true;
 					$$ = (Node *)n;
 				}
+			/* ALTER TABLE <name> ALTER [COLUMN] <colname> SET (COMPRESSION <cm> [WITH (<options>)]) */
+			| ALTER opt_column ColId SET alterColumnCompression
+				{
+					AlterTableCmd *n = makeNode(AlterTableCmd);
+					n->subtype = AT_SetCompression;
+					n->name = $3;
+					n->def = $5;
+					$$ = (Node *)n;
+				}
 			/* ALTER TABLE <name> DROP [COLUMN] IF EXISTS <colname> [RESTRICT|CASCADE] */
 			| DROP opt_column IF_P EXISTS ColId opt_drop_behavior
 				{
@@ -3381,7 +3393,7 @@ columnDef:	ColId Typename optColumnCompression create_generic_options ColQualLis
 					ColumnDef *n = makeNode(ColumnDef);
 					n->colname = $1;
 					n->typeName = $2;
-					n->compression = $3;
+					n->compression = (ColumnCompression *) $3;
 					n->inhcount = 0;
 					n->is_local = true;
 					n->is_not_null = false;
@@ -3436,13 +3448,42 @@ columnOptions:	ColId ColQualList
 				}
 		;
 
+optCompressionPreserve:
+			PRESERVE '(' name_list ')' { $$ = $3; }
+			| /*EMPTY*/ { $$ = NULL; }
+		;
+
+optCompressionParameters:
+			WITH '(' generic_option_list ')' { $$ = $3; }
+			| /*EMPTY*/	{ $$ = NULL; }
+		;
+
+compressionClause:
+			COMPRESSION name { $$ = pstrdup($2); }
+		;
+
 optColumnCompression:
-					COMPRESSION name
-					{
-						$$ = $2;
-					}
-					| /*EMPTY*/	{ $$ = NULL; }
-				;
+			compressionClause optCompressionParameters
+				{
+					ColumnCompression *n = makeNode(ColumnCompression);
+					n->amname = $1;
+					n->options = (List *) $2;
+					n->preserve = NIL;
+					$$ = (Node *) n;
+				}
+			| /*EMPTY*/	{ $$ = NULL; }
+		;
+
+alterColumnCompression:
+			compressionClause optCompressionParameters optCompressionPreserve
+				{
+					ColumnCompression *n = makeNode(ColumnCompression);
+					n->amname = $1;
+					n->options = (List *) $2;
+					n->preserve = (List *) $3;
+					$$ = (Node *) n;
+				}
+		;
 
 ColQualList:
 			ColQualList ColConstraint				{ $$ = lappend($1, $2); }
diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c
index 0638e56cd4..1bdb3ed1b1 100644
--- a/src/backend/parser/parse_utilcmd.c
+++ b/src/backend/parser/parse_utilcmd.c
@@ -1130,7 +1130,7 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
 		/* Likewise, copy compression if requested */
 		if (table_like_clause->options & CREATE_TABLE_LIKE_COMPRESSION
 			&& OidIsValid(attribute->attcompression))
-			def->compression = GetCompressionName(attribute->attcompression);
+			def->compression = MakeColumnCompression(attribute->attcompression);
 		else
 			def->compression = NULL;
 
diff --git a/src/backend/utils/adt/pg_upgrade_support.c b/src/backend/utils/adt/pg_upgrade_support.c
index 14d9eb2b5b..0ba8a14a56 100644
--- a/src/backend/utils/adt/pg_upgrade_support.c
+++ b/src/backend/utils/adt/pg_upgrade_support.c
@@ -105,6 +105,16 @@ binary_upgrade_set_next_pg_authid_oid(PG_FUNCTION_ARGS)
 	PG_RETURN_VOID();
 }
 
+Datum
+binary_upgrade_set_next_attr_compression_oid(PG_FUNCTION_ARGS)
+{
+	Oid			acoid = PG_GETARG_OID(0);
+
+	CHECK_IS_BINARY_UPGRADE;
+	binary_upgrade_next_attr_compression_oid = acoid;
+	PG_RETURN_VOID();
+}
+
 Datum
 binary_upgrade_create_empty_extension(PG_FUNCTION_ARGS)
 {
diff --git a/src/backend/utils/cache/syscache.c b/src/backend/utils/cache/syscache.c
index 53d9ddf159..dc4c672250 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"
@@ -188,6 +189,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/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 2eeeea306f..6610a340bf 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -45,6 +45,7 @@
 #include "catalog/pg_aggregate_d.h"
 #include "catalog/pg_am_d.h"
 #include "catalog/pg_attribute_d.h"
+#include "catalog/pg_attr_compression_d.h"
 #include "catalog/pg_cast_d.h"
 #include "catalog/pg_class_d.h"
 #include "catalog/pg_default_acl_d.h"
@@ -8529,9 +8530,16 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 
 		if (createWithCompression)
 			appendPQExpBuffer(q,
-							  "c.amname AS attcmname,\n");
+							  "pg_catalog.array_to_string(ARRAY("
+							  "SELECT pg_catalog.quote_ident(option_name) || "
+							  "' ' || pg_catalog.quote_literal(option_value) "
+							  "FROM pg_catalog.pg_options_to_table(c.acoptions) "
+							  "ORDER BY option_name"
+							  "), E',\n    ') AS attcmoptions,\n"
+							  "c.acname AS attcmname,\n");
 		else
 			appendPQExpBuffer(q,
+							  "NULL AS attcmoptions,\n"
 							  "NULL AS attcmname,\n");
 
 
@@ -8556,8 +8564,8 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 						  "ON a.atttypid = t.oid\n");
 
 		if (createWithCompression)
-			appendPQExpBuffer(q, "LEFT JOIN pg_catalog.pg_am c "
-								 "ON a.attcompression = c.oid\n");
+			appendPQExpBuffer(q, "LEFT JOIN pg_catalog.pg_attr_compression c "
+								 "ON a.attcompression = c.acoid\n");
 
 		appendPQExpBuffer(q,
 						  "WHERE a.attrelid = '%u'::pg_catalog.oid "
@@ -8586,10 +8594,12 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 		tbinfo->attcollation = (Oid *) pg_malloc(ntups * sizeof(Oid));
 		tbinfo->attfdwoptions = (char **) pg_malloc(ntups * sizeof(char *));
 		tbinfo->attmissingval = (char **) pg_malloc(ntups * sizeof(char *));
+		tbinfo->attcmoptions = (char **) pg_malloc(ntups * sizeof(char *));
 		tbinfo->attcmnames = (char **) pg_malloc(ntups * sizeof(char *));
 		tbinfo->notnull = (bool *) pg_malloc(ntups * sizeof(bool));
 		tbinfo->inhNotNull = (bool *) pg_malloc(ntups * sizeof(bool));
 		tbinfo->attrdefs = (AttrDefInfo **) pg_malloc(ntups * sizeof(AttrDefInfo *));
+		tbinfo->attcompression = NULL;
 		hasdefaults = false;
 
 		for (int j = 0; j < ntups; j++)
@@ -8616,6 +8626,7 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 			tbinfo->attfdwoptions[j] = pg_strdup(PQgetvalue(res, j, PQfnumber(res, "attfdwoptions")));
 			tbinfo->attmissingval[j] = pg_strdup(PQgetvalue(res, j, PQfnumber(res, "attmissingval")));
 			tbinfo->attcmnames[j] = pg_strdup(PQgetvalue(res, j, PQfnumber(res, "attcmname")));
+			tbinfo->attcmoptions[j] = pg_strdup(PQgetvalue(res, j, PQfnumber(res, "attcmoptions")));
 			tbinfo->attrdefs[j] = NULL; /* fix below */
 			tbinfo->attrdefs[j] = NULL; /* fix below */
 			if (PQgetvalue(res, j, PQfnumber(res, "atthasdef"))[0] == 't')
@@ -8831,6 +8842,104 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables)
 			}
 			PQclear(res);
 		}
+
+		/*
+		 * Get compression info
+		 */
+		if (fout->remoteVersion >= 120000 && dopt->binary_upgrade)
+		{
+			int			i_acname;
+			int			i_acoid;
+			int			i_parsedoptions;
+			int			i_curattnum;
+			int			start;
+
+			pg_log_info("finding compression info for table \"%s.%s\"",
+						tbinfo->dobj.namespace->dobj.name,
+						tbinfo->dobj.name);
+
+			tbinfo->attcompression = pg_malloc0(tbinfo->numatts * sizeof(AttrCompressionInfo *));
+
+			resetPQExpBuffer(q);
+			appendPQExpBuffer(q,
+				"SELECT attrelid::pg_catalog.regclass AS relname, attname,"
+				" (CASE WHEN deptype = 'i' THEN refobjsubid ELSE objsubid END) AS curattnum,"
+				" (CASE WHEN deptype = 'n' THEN attcompression = refobjid"
+				"		ELSE attcompression = objid END) AS iscurrent,"
+				" acname, acoid,"
+				" (CASE WHEN acoptions IS NOT NULL"
+				"  THEN pg_catalog.array_to_string(ARRAY("
+				"		SELECT pg_catalog.quote_ident(option_name) || "
+				"			' ' || pg_catalog.quote_literal(option_value) "
+				"		FROM pg_catalog.pg_options_to_table(acoptions) "
+				"		ORDER BY option_name"
+				"		), E',\n    ')"
+				"  ELSE NULL END) AS parsedoptions "
+				" FROM pg_depend d"
+				" JOIN pg_attribute a ON"
+				"	(classid = 'pg_class'::pg_catalog.regclass::pg_catalog.oid AND a.attrelid = d.objid"
+				"		AND a.attnum = d.objsubid AND d.deptype = 'n'"
+				"		AND d.refclassid = 'pg_attr_compression'::pg_catalog.regclass::pg_catalog.oid)"
+				"	OR (d.refclassid = 'pg_class'::pg_catalog.regclass::pg_catalog.oid"
+				"		AND d.refobjid = a.attrelid"
+				"		AND d.refobjsubid = a.attnum AND d.deptype = 'i'"
+				"		AND d.classid = 'pg_attr_compression'::pg_catalog.regclass::pg_catalog.oid)"
+				" JOIN pg_attr_compression c ON"
+				"	(d.deptype = 'i' AND d.objid = c.acoid AND a.attnum = c.acattnum"
+				"		AND a.attrelid = c.acrelid) OR"
+				"	(d.deptype = 'n' AND d.refobjid = c.acoid AND c.acattnum = 0"
+				"		AND c.acrelid = 0)"
+				" WHERE (deptype = 'n' AND d.objid = %d) OR (deptype = 'i' AND d.refobjid = %d)"
+				" ORDER BY curattnum, iscurrent;",
+				tbinfo->dobj.catId.oid, tbinfo->dobj.catId.oid);
+
+			res = ExecuteSqlQuery(fout, q->data, PGRES_TUPLES_OK);
+			ntups = PQntuples(res);
+
+			if (ntups > 0)
+			{
+				int		j;
+				int		k;
+
+				i_acname = PQfnumber(res, "acname");
+				i_acoid = PQfnumber(res, "acoid");
+				i_parsedoptions = PQfnumber(res, "parsedoptions");
+				i_curattnum = PQfnumber(res, "curattnum");
+
+				start = 0;
+
+				for (j = 0; j < ntups; j++)
+				{
+					int		attnum = atoi(PQgetvalue(res, j, i_curattnum));
+
+					if ((j == ntups - 1) || atoi(PQgetvalue(res, j + 1, i_curattnum)) != attnum)
+					{
+						AttrCompressionInfo *cminfo = pg_malloc(sizeof(AttrCompressionInfo));
+
+						cminfo->nitems = j - start + 1;
+						cminfo->items = pg_malloc(sizeof(AttrCompressionItem *) * cminfo->nitems);
+
+						for (k = start; k < start + cminfo->nitems; k++)
+						{
+							AttrCompressionItem	*cmitem = pg_malloc0(sizeof(AttrCompressionItem));
+
+							cmitem->acname = pg_strdup(PQgetvalue(res, k, i_acname));
+							cmitem->acoid = atooid(PQgetvalue(res, k, i_acoid));
+
+							if (!PQgetisnull(res, k, i_parsedoptions))
+								cmitem->parsedoptions = pg_strdup(PQgetvalue(res, k, i_parsedoptions));
+
+							cminfo->items[k - start] = cmitem;
+						}
+
+						tbinfo->attcompression[attnum - 1] = cminfo;
+						start = j + 1;	/* start from next */
+					}
+				}
+			}
+
+			PQclear(res);
+		}
 	}
 
 	destroyPQExpBuffer(q);
@@ -15694,7 +15803,8 @@ dumpTableSchema(Archive *fout, TableInfo *tbinfo)
 					 * pg_attr_compression
 					 */
 					has_custom_compression = (tbinfo->attcmnames[j] &&
-						((strcmp(tbinfo->attcmnames[j], "pglz") != 0)));
+						((strcmp(tbinfo->attcmnames[j], "pglz") != 0) ||
+						 nonemptyReloptions(tbinfo->attcmoptions[j])));
 
 					/* Format properly if not first attr */
 					if (actual_atts == 0)
@@ -15745,6 +15855,9 @@ dumpTableSchema(Archive *fout, TableInfo *tbinfo)
 					{
 						appendPQExpBuffer(q, " COMPRESSION %s",
 										  tbinfo->attcmnames[j]);
+						if (nonemptyReloptions(tbinfo->attcmoptions[j]))
+							appendPQExpBuffer(q, " WITH (%s)",
+											  tbinfo->attcmoptions[j]);
 					}
 
 					if (print_default)
@@ -16186,6 +16299,33 @@ dumpTableSchema(Archive *fout, TableInfo *tbinfo)
 								  qualrelname,
 								  fmtId(tbinfo->attnames[j]),
 								  tbinfo->attfdwoptions[j]);
+			/*
+			 * Dump per-column compression options
+			 */
+			if (tbinfo->attcompression && tbinfo->attcompression[j])
+			{
+				AttrCompressionInfo *cminfo = tbinfo->attcompression[j];
+
+				if (cminfo->nitems)
+					appendPQExpBuffer(q, "\n-- For binary upgrade, recreate compression metadata on column %s\n",
+							fmtId(tbinfo->attnames[j]));
+
+				for (int i = 0; i < cminfo->nitems; i++)
+				{
+					AttrCompressionItem *item = cminfo->items[i];
+
+					appendPQExpBuffer(q,
+						"SELECT binary_upgrade_set_next_attr_compression_oid('%d'::pg_catalog.oid);\n",
+									  item->acoid);
+					appendPQExpBuffer(q, "ALTER TABLE %s ALTER COLUMN %s\nSET COMPRESSION %s",
+									  qualrelname, fmtId(tbinfo->attnames[j]), item->acname);
+
+					if (item->parsedoptions)
+						appendPQExpBuffer(q, "\nWITH (%s);\n", item->parsedoptions);
+					else
+						appendPQExpBuffer(q, ";\n");
+				}
+			}
 		}
 
 		if (ftoptions)
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index f9c87e6c16..a8891cb3f5 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -325,7 +325,9 @@ typedef struct _tableInfo
 	bool		needs_override; /* has GENERATED ALWAYS AS IDENTITY */
 	char	   *amname;			/* relation access method */
 
+	char	  **attcmoptions;	/* per-attribute current compression options */
 	char	  **attcmnames;		/* per-attribute current compression method names */
+	struct _attrCompressionInfo **attcompression; /* per-attribute all compression data */
 
 	/*
 	 * Stuff computed only for dumpable tables.
@@ -348,6 +350,19 @@ typedef struct _attrDefInfo
 	bool		separate;		/* true if must dump as separate item */
 } AttrDefInfo;
 
+typedef struct _attrCompressionItem
+{
+	Oid			acoid;			/* attribute compression oid */
+	char	   *acname;			/* compression access method name */
+	char	   *parsedoptions;	/* WITH options */
+} AttrCompressionItem;
+
+typedef struct _attrCompressionInfo
+{
+	int			nitems;
+	AttrCompressionItem	**items;
+} AttrCompressionInfo;
+
 typedef struct _tableDataInfo
 {
 	DumpableObject dobj;
diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl
index ec63662060..be0c14883b 100644
--- a/src/bin/pg_dump/t/002_pg_dump.pl
+++ b/src/bin/pg_dump/t/002_pg_dump.pl
@@ -657,6 +657,43 @@ my %tests = (
 		},
 	},
 
+	# compression data in binary upgrade mode
+	'ALTER TABLE test_table_compression ALTER COLUMN ... SET COMPRESSION' => {
+		all_runs  => 1,
+		catch_all => 'ALTER TABLE ... commands',
+		regexp    => qr/^
+			\QCREATE TABLE dump_test.test_table_compression (\E\n
+			\s+\Qcol1 text,\E\n
+			\s+\Qcol2 text,\E\n
+			\s+\Qcol3 text,\E\n
+			\s+\Qcol4 text\E\n
+			\);
+			.*
+			\QSELECT binary_upgrade_set_next_attr_compression_oid('\E\d+\Q'::pg_catalog.oid);\E\n
+			\QALTER TABLE dump_test.test_table_compression ALTER COLUMN col1\E\n
+			\QSET COMPRESSION pglz;\E\n
+			.*
+			\QSELECT binary_upgrade_set_next_attr_compression_oid('\E\d+\Q'::pg_catalog.oid);\E\n
+			\QALTER TABLE dump_test.test_table_compression ALTER COLUMN col2\E\n
+			\QSET COMPRESSION pglz2;\E\n
+			.*
+			\QSELECT binary_upgrade_set_next_attr_compression_oid('\E\d+\Q'::pg_catalog.oid);\E\n
+			\QALTER TABLE dump_test.test_table_compression ALTER COLUMN col3\E\n
+			\QSET COMPRESSION pglz\E\n
+			\QWITH (min_input_size '1000');\E\n
+			.*
+			\QSELECT binary_upgrade_set_next_attr_compression_oid('\E\d+\Q'::pg_catalog.oid);\E\n
+			\QALTER TABLE dump_test.test_table_compression ALTER COLUMN col4\E\n
+			\QSET COMPRESSION pglz2\E\n
+			\QWITH (min_input_size '1000');\E\n
+			\QSELECT binary_upgrade_set_next_attr_compression_oid('\E\d+\Q'::pg_catalog.oid);\E\n
+			\QALTER TABLE dump_test.test_table_compression ALTER COLUMN col4\E\n
+			\QSET COMPRESSION pglz2\E\n
+			\QWITH (min_input_size '2000');\E\n
+			/xms,
+		like => { binary_upgrade => 1, },
+	},
+
 	'ALTER TABLE ONLY test_table ALTER COLUMN col1 SET STATISTICS 90' => {
 		create_order => 93,
 		create_sql =>
@@ -1379,6 +1416,17 @@ my %tests = (
 		like => { %full_runs, section_pre_data => 1, },
 	},
 
+	'CREATE ACCESS METHOD pglz2' => {
+		all_runs     => 1,
+		catch_all    => 'CREATE ... commands',
+		create_order => 52,
+		create_sql =>
+		  'CREATE ACCESS METHOD pglz2 TYPE COMPRESSION HANDLER pglzhandler;',
+		regexp =>
+		  qr/CREATE ACCESS METHOD pglz2 TYPE COMPRESSION HANDLER pglzhandler;/m,
+		like => { %full_runs, section_pre_data => 1, },
+	},
+
 	'CREATE COLLATION test0 FROM "C"' => {
 		create_order => 76,
 		create_sql   => 'CREATE COLLATION test0 FROM "C";',
@@ -2553,6 +2601,53 @@ my %tests = (
 		},
 	},
 
+	'CREATE TABLE test_table_compression' => {
+		create_order => 55,
+		create_sql   => 'CREATE TABLE dump_test.test_table_compression (
+						   col1 text,
+						   col2 text COMPRESSION pglz2,
+						   col3 text COMPRESSION pglz WITH (min_input_size \'1000\'),
+						   col4 text COMPRESSION pglz2 WITH (min_input_size \'1000\')
+					     );',
+		regexp => qr/^
+			\QCREATE TABLE dump_test.test_table_compression (\E\n
+			\s+\Qcol1 text,\E\n
+			\s+\Qcol2 text COMPRESSION pglz2,\E\n
+			\s+\Qcol3 text COMPRESSION pglz WITH (min_input_size '1000'),\E\n
+			\s+\Qcol4 text COMPRESSION pglz2 WITH (min_input_size '2000')\E\n
+			\);
+			/xm,
+		like =>
+		  { %full_runs, %dump_test_schema_runs, section_pre_data => 1, },
+		unlike => {
+			binary_upgrade		     => 1,
+			exclude_dump_test_schema => 1,
+		},
+	},
+
+	'ALTER TABLE test_table_compression' => {
+		create_order => 56,
+		create_sql   => 'ALTER TABLE dump_test.test_table_compression
+						 ALTER COLUMN col4
+						 SET COMPRESSION pglz2
+						 WITH (min_input_size \'2000\')
+						 PRESERVE (pglz2);',
+		regexp => qr/^
+			\QCREATE TABLE dump_test.test_table_compression (\E\n
+			\s+\Qcol1 text,\E\n
+			\s+\Qcol2 text COMPRESSION pglz2,\E\n
+			\s+\Qcol3 text COMPRESSION pglz WITH (min_input_size '1000'),\E\n
+			\s+\Qcol4 text COMPRESSION pglz2 WITH (min_input_size '2000')\E\n
+			\);
+			/xm,
+		like =>
+		  { %full_runs, %dump_test_schema_runs, section_pre_data => 1, },
+		unlike => {
+			binary_upgrade		     => 1,
+			exclude_dump_test_schema => 1,
+		},
+	},
+
 	'CREATE STATISTICS extended_stats_no_options' => {
 		create_order => 97,
 		create_sql   => 'CREATE STATISTICS dump_test.test_ext_stats_no_options
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index 5a511865ec..887d4d3491 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -1895,9 +1895,14 @@ describeOneTableDetails(const char *schemaname,
 			 tableinfo.relkind == RELKIND_PARTITIONED_TABLE))
 		{
 			appendPQExpBufferStr(&buf, ",\n  CASE WHEN attcompression = 0 THEN NULL ELSE "
-								 " (SELECT c.amname "
-								 "  FROM pg_catalog.pg_am c "
-								 "  WHERE c.oid = a.attcompression) "
+								 " (SELECT c.acname || "
+								 "		(CASE WHEN acoptions IS NULL "
+								 "		 THEN '' "
+								 "		 ELSE '(' || array_to_string(ARRAY(SELECT quote_ident(option_name) || ' ' || quote_literal(option_value)"
+								 "											  FROM pg_options_to_table(acoptions)), ', ') || ')'"
+								 " 		 END) "
+								 "  FROM pg_catalog.pg_attr_compression c "
+								 "  WHERE c.acoid = a.attcompression) "
 								 " END AS attcmname");
 			attcompression_col = cols++;
 		}
diff --git a/src/include/access/cmapi.h b/src/include/access/cmapi.h
index c9b074deea..d9f2871712 100644
--- a/src/include/access/cmapi.h
+++ b/src/include/access/cmapi.h
@@ -14,12 +14,12 @@
 #define CMAPI_H
 
 #include "postgres.h"
-#include "catalog/pg_am.h"
+#include "catalog/pg_attr_compression.h"
 #include "catalog/pg_attribute.h"
 #include "nodes/pg_list.h"
 
 #define IsBuiltinCompression(cmid)	((cmid) < FirstBootstrapObjectId)
-#define DefaultCompressionOid		(PGLZ_COMPRESSION_AM_OID)
+#define DefaultCompressionOid		(PGLZ_AC_OID)
 
 typedef struct CompressionAmRoutine CompressionAmRoutine;
 
@@ -32,18 +32,24 @@ typedef struct CompressionAmRoutine CompressionAmRoutine;
  */
 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 struct varlena *(*cmdecompress_slice_function)
 			(CompressionAmOptions *cmoptions, const struct varlena *value,
 			 int32 slicelength);
-typedef void *(*cminitstate_function) (Oid acoid);
+typedef void *(*cminitstate_function) (Oid acoid, List *options);
 
 /*
  * API struct for a compression AM.
@@ -63,6 +69,7 @@ struct CompressionAmRoutine
 {
 	NodeTag		type;
 
+	cmcheck_function cmcheck;	/* can be NULL */
 	cminitstate_function cminitstate;	/* can be NULL */
 	cmcompress_function cmcompress;
 	cmcompress_function cmdecompress;
diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h
index ba77013f64..4c89a2cfa9 100644
--- a/src/include/access/heapam.h
+++ b/src/include/access/heapam.h
@@ -28,6 +28,7 @@
 #include "storage/shm_toc.h"
 #include "utils/relcache.h"
 #include "utils/snapshot.h"
+#include "utils/hsearch.h"
 
 
 /* "options" flag bits for heap_insert */
@@ -131,7 +132,7 @@ extern bool heap_hot_search_buffer(ItemPointer tid, Relation relation,
 extern void heap_get_latest_tid(TableScanDesc scan, ItemPointer tid);
 extern void setLastTid(const ItemPointer tid);
 
-extern BulkInsertState GetBulkInsertState(void);
+extern BulkInsertState GetBulkInsertState(HTAB *);
 extern void FreeBulkInsertState(BulkInsertState);
 extern void ReleaseBulkInsertStatePin(BulkInsertState bistate);
 
diff --git a/src/include/access/heaptoast.h b/src/include/access/heaptoast.h
index 7169254199..6d35aafa2d 100644
--- a/src/include/access/heaptoast.h
+++ b/src/include/access/heaptoast.h
@@ -96,7 +96,8 @@
  * ----------
  */
 extern HeapTuple heap_toast_insert_or_update(Relation rel, HeapTuple newtup,
-											 HeapTuple oldtup, int options);
+											 HeapTuple oldtup, int options,
+											 HTAB *preserved_am_info);
 
 /* ----------
  * heap_toast_delete -
diff --git a/src/include/access/hio.h b/src/include/access/hio.h
index f69a92521b..ad82fa7755 100644
--- a/src/include/access/hio.h
+++ b/src/include/access/hio.h
@@ -30,6 +30,8 @@ typedef struct BulkInsertStateData
 {
 	BufferAccessStrategy strategy;	/* our BULKWRITE strategy object */
 	Buffer		current_buf;	/* current insertion target page */
+	HTAB	   *preserved_am_info;	/* hash table with preserved compression
+									 * methods for attributes */
 } BulkInsertStateData;
 
 
diff --git a/src/include/access/toast_helper.h b/src/include/access/toast_helper.h
index 4e932bc067..7d1f941b95 100644
--- a/src/include/access/toast_helper.h
+++ b/src/include/access/toast_helper.h
@@ -14,8 +14,10 @@
 #ifndef TOAST_HELPER_H
 #define TOAST_HELPER_H
 
+#include "utils/hsearch.h"
 #include "utils/rel.h"
 
+
 /*
  * Information about one column of a tuple being toasted.
  *
@@ -101,7 +103,7 @@ typedef struct
 #define TOASTCOL_IGNORE						0x0010
 #define TOASTCOL_INCOMPRESSIBLE				0x0020
 
-extern void toast_tuple_init(ToastTupleContext *ttc);
+extern void toast_tuple_init(ToastTupleContext *ttc, HTAB *preserved_am_info);
 extern int	toast_tuple_find_biggest_attribute(ToastTupleContext *ttc,
 											   bool for_compression,
 											   bool check_main);
diff --git a/src/include/access/toast_internals.h b/src/include/access/toast_internals.h
index 94e11e583f..ef16034699 100644
--- a/src/include/access/toast_internals.h
+++ b/src/include/access/toast_internals.h
@@ -34,9 +34,22 @@ 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_am */
+	Oid cmid;	   /* Oid from pg_attr_compression */
 } toast_compress_header_custom;
 
+/*
+ * 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;
+
+HTAB *amoptions_cache;
+MemoryContext amoptions_cache_mcxt;
+
 #define RAWSIZEMASK (0x3FFFFFFFU)
 
 /*
@@ -116,14 +129,14 @@ extern int	toast_open_indexes(Relation toastrel,
 extern void toast_close_indexes(Relation *toastidxs, int num_indexes,
 								LOCKMODE lock);
 extern void init_toast_snapshot(Snapshot toast_snapshot);
-
+extern void init_amoptions_cache(void);
+extern bool attr_compression_options_are_equal(Oid acoid1, Oid acoid2);
 /*
  * lookup_compression_am_options -
  *
  * Return cached CompressionAmOptions for specified attribute compression.
  */
-extern void lookup_compression_am_options(Oid acoid,
-										 CompressionAmOptions *result);
+extern CompressionAmOptions *lookup_compression_am_options(Oid acoid);
 
 /*
  * toast_set_compressed_datum_info -
diff --git a/src/include/catalog/binary_upgrade.h b/src/include/catalog/binary_upgrade.h
index 02fecb90f7..8b80dd6850 100644
--- a/src/include/catalog/binary_upgrade.h
+++ b/src/include/catalog/binary_upgrade.h
@@ -24,6 +24,8 @@ extern PGDLLIMPORT Oid binary_upgrade_next_toast_pg_class_oid;
 extern PGDLLIMPORT Oid binary_upgrade_next_pg_enum_oid;
 extern PGDLLIMPORT Oid binary_upgrade_next_pg_authid_oid;
 
+extern PGDLLIMPORT Oid binary_upgrade_next_attr_compression_oid;
+
 extern PGDLLIMPORT bool binary_upgrade_record_init_privs;
 
 #endif							/* BINARY_UPGRADE_H */
diff --git a/src/include/catalog/dependency.h b/src/include/catalog/dependency.h
index a8f7e9965b..e35ddd64d8 100644
--- a/src/include/catalog/dependency.h
+++ b/src/include/catalog/dependency.h
@@ -125,10 +125,11 @@ typedef enum ObjectClass
 	OCLASS_PUBLICATION,			/* pg_publication */
 	OCLASS_PUBLICATION_REL,		/* pg_publication_rel */
 	OCLASS_SUBSCRIPTION,		/* pg_subscription */
-	OCLASS_TRANSFORM			/* pg_transform */
+	OCLASS_TRANSFORM,			/* pg_transform */
+	OCLASS_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 a7e2a9b26b..e1448dd7fd 100644
--- a/src/include/catalog/indexing.h
+++ b/src/include/catalog/indexing.h
@@ -95,6 +95,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, 2137, on pg_attr_compression using btree(acoid oid_ops));
+#define AttrCompressionIndexId 2137
+DECLARE_INDEX(pg_attr_compression_relid_attnum_index, 2121, on pg_attr_compression using btree(acrelid oid_ops, acattnum int2_ops));
+#define AttrCompressionRelidAttnumIndexId  2121
+
 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..f0683083ed
--- /dev/null
+++ b/src/include/catalog/pg_attr_compression.dat
@@ -0,0 +1,24 @@
+#----------------------------------------------------------------------
+#
+# 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
+#
+#----------------------------------------------------------------------
+
+[
+
+{ acoid => '4225', acname => 'pglz' },
+{ acoid => '4226', acname => 'zlib' },
+
+]
diff --git a/src/include/catalog/pg_attr_compression.h b/src/include/catalog/pg_attr_compression.h
new file mode 100644
index 0000000000..d6550a0eec
--- /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,5555,AttrCompressionRelationId)
+{
+	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_proc.dat b/src/include/catalog/pg_proc.dat
index 9404a24199..902a9fb78c 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -7022,6 +7022,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 => '2228', 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',
@@ -10366,6 +10369,10 @@
   proname => 'binary_upgrade_set_missing_value', provolatile => 'v',
   proparallel => 'u', prorettype => 'void', proargtypes => 'oid text text',
   prosrc => 'binary_upgrade_set_missing_value' },
+{ oid => '4035', descr => 'for use by pg_upgrade',
+  proname => 'binary_upgrade_set_next_attr_compression_oid', provolatile => 'v',
+  proparallel => 'r', prorettype => 'void', proargtypes => 'oid',
+  prosrc => 'binary_upgrade_set_next_attr_compression_oid' },
 
 # conversion functions
 { oid => '4302',
diff --git a/src/include/catalog/toasting.h b/src/include/catalog/toasting.h
index 51491c4513..8370509822 100644
--- a/src/include/catalog/toasting.h
+++ b/src/include/catalog/toasting.h
@@ -75,6 +75,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, 5556, 5558);
 
 /* shared catalogs */
 DECLARE_TOAST(pg_authid, 4175, 4176);
diff --git a/src/include/commands/defrem.h b/src/include/commands/defrem.h
index 8d0de0babe..0f533314d9 100644
--- a/src/include/commands/defrem.h
+++ b/src/include/commands/defrem.h
@@ -133,6 +133,7 @@ extern ObjectAddress AlterUserMapping(AlterUserMappingStmt *stmt);
 extern Oid	RemoveUserMapping(DropUserMappingStmt *stmt);
 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,
@@ -148,8 +149,15 @@ extern char *get_am_name(Oid amOid);
 extern regproc get_am_handler_oid(Oid amOid, char amtype, bool noerror);
 
 /* commands/compressioncmds.c */
-extern char *GetCompressionName(Oid acoid);
-extern Oid GetAttributeCompression(Form_pg_attribute attr, char *compression);
+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 35df26f978..205ae1210f 100644
--- a/src/include/nodes/nodes.h
+++ b/src/include/nodes/nodes.h
@@ -480,6 +480,7 @@ typedef enum NodeTag
 	T_PartitionBoundSpec,
 	T_PartitionRangeDatum,
 	T_PartitionCmd,
+	T_ColumnCompression,
 	T_VacuumRelation,
 
 	/*
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 8182f8edba..8a101466bd 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -623,6 +623,20 @@ typedef struct RangeTableSample
 	int			location;		/* method name location, or -1 if unknown */
 } RangeTableSample;
 
+/*
+ * ColumnCompression - compression parameters for some attribute
+ *
+ * This represents compression information defined using clause:
+ * .. COMPRESSION <compression method> WITH (<params>) PRESERVE <compression AMs>
+ */
+typedef struct ColumnCompression
+{
+	NodeTag		type;
+	char	   *amname;
+	List	   *options;
+	List	   *preserve;
+} ColumnCompression;
+
 /*
  * ColumnDef - column definition (used in various creates)
  *
@@ -646,7 +660,7 @@ typedef struct ColumnDef
 	NodeTag		type;
 	char	   *colname;		/* name of column */
 	TypeName   *typeName;		/* type of column */
-	char	   *compression;
+	ColumnCompression *compression;
 	int			inhcount;		/* number of times column is inherited */
 	bool		is_local;		/* column has local (non-inherited) def'n */
 	bool		is_not_null;	/* NOT NULL constraint specified? */
@@ -1849,7 +1863,8 @@ typedef enum AlterTableType
 	AT_DetachPartition,			/* DETACH PARTITION */
 	AT_AddIdentity,				/* ADD IDENTITY */
 	AT_SetIdentity,				/* SET identity column options */
-	AT_DropIdentity				/* DROP IDENTITY */
+	AT_DropIdentity,			/* DROP IDENTITY */
+	AT_SetCompression			/* SET COMPRESSION */
 } AlterTableType;
 
 typedef struct ReplicaIdentityStmt
diff --git a/src/include/utils/syscache.h b/src/include/utils/syscache.h
index f27b73d76d..b5c2f5da51 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/test/regress/expected/create_cm.out b/src/test/regress/expected/create_cm.out
index ee091bb01f..d3b4095fea 100644
--- a/src/test/regress/expected/create_cm.out
+++ b/src/test/regress/expected/create_cm.out
@@ -1,53 +1,273 @@
+-- test drop
+DROP ACCESS METHOD pglz; --fail
+ERROR:  cannot drop access method pglz because it is required by the database system
+CREATE ACCESS METHOD pglz1 TYPE COMPRESSION HANDLER pglzhandler;
+CREATE TABLE droptest(d1 TEXT COMPRESSION pglz1);
+DROP ACCESS METHOD pglz1;
+ERROR:  cannot drop access method pglz1 because other objects depend on it
+DETAIL:  column d1 of table droptest depends on access method pglz1
+HINT:  Use DROP ... CASCADE to drop the dependent objects too.
+DROP ACCESS METHOD pglz1 CASCADE;
+NOTICE:  drop cascades to column d1 of table droptest
+\d+ droptest
+                                       Table "public.droptest"
+ Column | Type | Collation | Nullable | Default | Storage | Compression | Stats target | Description 
+--------+------+-----------+----------+---------+---------+-------------+--------------+-------------
+
+DROP TABLE droptest;
+CREATE ACCESS METHOD pglz1 TYPE COMPRESSION HANDLER pglzhandler;
+-- test auto drop of related attribute compression
+CREATE ACCESS METHOD pglz2 TYPE COMPRESSION HANDLER pglzhandler;
+CREATE TABLE cmaltertest (f1 TEXT COMPRESSION pglz2);
+SELECT acname, acattnum, acoptions FROM pg_attr_compression
+	WHERE acrelid = 'cmaltertest'::REGCLASS;
+ acname | acattnum | acoptions 
+--------+----------+-----------
+ pglz2  |        1 | 
+(1 row)
+
+ALTER TABLE cmaltertest DROP COLUMN f1;
+SELECT acname, acattnum, acoptions FROM pg_attr_compression
+	WHERE acrelid = 'cmaltertest'::REGCLASS;
+ acname | acattnum | acoptions 
+--------+----------+-----------
+(0 rows)
+
+DROP TABLE cmaltertest;
+-- test drop
+DROP ACCESS METHOD pglz2;
+-- test alter data type
+CREATE TABLE cmaltertest(at1 TEXT);
+ALTER TABLE cmaltertest ALTER COLUMN at1 SET COMPRESSION pglz1 PRESERVE (pglz);
+SELECT pg_column_compression('cmaltertest', 'at1');
+ pg_column_compression 
+-----------------------
+ pglz, pglz1
+(1 row)
+
+ALTER TABLE cmaltertest ALTER COLUMN at1 SET DATA TYPE INT USING at1::INTEGER;
+\d+ cmaltertest
+                                       Table "public.cmaltertest"
+ Column |  Type   | Collation | Nullable | Default | Storage | Compression | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+-------------+--------------+-------------
+ at1    | integer |           |          |         | plain   |             |              | 
+
+SELECT pg_column_compression('cmaltertest', 'at1');
+ pg_column_compression 
+-----------------------
+ 
+(1 row)
+
+SELECT acname, acattnum, acoptions FROM pg_attr_compression
+	WHERE acrelid = 'cmaltertest'::REGCLASS;
+ acname | acattnum | acoptions 
+--------+----------+-----------
+(0 rows)
+
+ALTER TABLE cmaltertest ALTER COLUMN at1 SET DATA TYPE TEXT;
+SELECT pg_column_compression('cmaltertest', 'at1');
+ pg_column_compression 
+-----------------------
+ pglz
+(1 row)
+
+DROP TABLE cmaltertest;
 -- test storages
 CREATE TABLE cmstoragetest(st1 TEXT, st2 INT);
+ALTER TABLE cmstoragetest ALTER COLUMN st2
+	SET COMPRESSION pglz WITH (min_input_size '100'); -- fail
+ERROR:  column data type integer does not support compression
+ALTER TABLE cmstoragetest ALTER COLUMN st1
+	SET COMPRESSION pglz WITH (min_input_size '100', min_comp_rate '50');
 ALTER TABLE cmstoragetest ALTER COLUMN st1 SET STORAGE EXTERNAL;
 \d+ cmstoragetest
-                                      Table "public.cmstoragetest"
- Column |  Type   | Collation | Nullable | Default | Storage  | Compression | Stats target | Description 
---------+---------+-----------+----------+---------+----------+-------------+--------------+-------------
- st1    | text    |           |          |         | external | pglz        |              | 
- st2    | integer |           |          |         | plain    |             |              | 
+                                                        Table "public.cmstoragetest"
+ Column |  Type   | Collation | Nullable | Default | Storage  |                  Compression                   | Stats target | Description 
+--------+---------+-----------+----------+---------+----------+------------------------------------------------+--------------+-------------
+ st1    | text    |           |          |         | external | pglz(min_comp_rate '50', min_input_size '100') |              | 
+ st2    | integer |           |          |         | plain    |                                                |              | 
 
 ALTER TABLE cmstoragetest ALTER COLUMN st1 SET STORAGE MAIN;
 \d+ cmstoragetest
-                                      Table "public.cmstoragetest"
- Column |  Type   | Collation | Nullable | Default | Storage | Compression | Stats target | Description 
---------+---------+-----------+----------+---------+---------+-------------+--------------+-------------
- st1    | text    |           |          |         | main    | pglz        |              | 
- st2    | integer |           |          |         | plain   |             |              | 
+                                                       Table "public.cmstoragetest"
+ Column |  Type   | Collation | Nullable | Default | Storage |                  Compression                   | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+------------------------------------------------+--------------+-------------
+ st1    | text    |           |          |         | main    | pglz(min_comp_rate '50', min_input_size '100') |              | 
+ st2    | integer |           |          |         | plain   |                                                |              | 
 
 ALTER TABLE cmstoragetest ALTER COLUMN st1 SET STORAGE PLAIN;
 \d+ cmstoragetest
-                                      Table "public.cmstoragetest"
- Column |  Type   | Collation | Nullable | Default | Storage | Compression | Stats target | Description 
---------+---------+-----------+----------+---------+---------+-------------+--------------+-------------
- st1    | text    |           |          |         | plain   | pglz        |              | 
- st2    | integer |           |          |         | plain   |             |              | 
+                                                       Table "public.cmstoragetest"
+ Column |  Type   | Collation | Nullable | Default | Storage |                  Compression                   | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+------------------------------------------------+--------------+-------------
+ st1    | text    |           |          |         | plain   | pglz(min_comp_rate '50', min_input_size '100') |              | 
+ st2    | integer |           |          |         | plain   |                                                |              | 
 
 ALTER TABLE cmstoragetest ALTER COLUMN st1 SET STORAGE EXTENDED;
 \d+ cmstoragetest
-                                      Table "public.cmstoragetest"
- Column |  Type   | Collation | Nullable | Default | Storage  | Compression | Stats target | Description 
---------+---------+-----------+----------+---------+----------+-------------+--------------+-------------
- st1    | text    |           |          |         | extended | pglz        |              | 
- st2    | integer |           |          |         | plain    |             |              | 
+                                                        Table "public.cmstoragetest"
+ Column |  Type   | Collation | Nullable | Default | Storage  |                  Compression                   | Stats target | Description 
+--------+---------+-----------+----------+---------+----------+------------------------------------------------+--------------+-------------
+ st1    | text    |           |          |         | extended | pglz(min_comp_rate '50', min_input_size '100') |              | 
+ st2    | integer |           |          |         | plain    |                                                |              | 
 
 DROP TABLE cmstoragetest;
-CREATE TABLE cmdata(f1 text COMPRESSION pglz);
+CREATE ACCESS METHOD pglz2 TYPE COMPRESSION HANDLER pglzhandler;
+-- test PRESERVE
+CREATE TABLE cmtest(f1 TEXT);
+\d+ cmtest
+                                        Table "public.cmtest"
+ Column | Type | Collation | Nullable | Default | Storage  | Compression | Stats target | Description 
+--------+------+-----------+----------+---------+----------+-------------+--------------+-------------
+ f1     | text |           |          |         | extended | pglz        |              | 
+
+-- view to check dependencies
+CREATE VIEW cmtest_deps AS
+	SELECT classid, objsubid, refclassid, refobjsubid, deptype
+	FROM pg_depend
+	WHERE (refclassid = 'pg_catalog.pg_attr_compression'::REGCLASS OR
+            classid = 'pg_catalog.pg_attr_compression'::REGCLASS) AND
+		  (objid = 'cmtest'::REGCLASS OR refobjid = 'cmtest'::REGCLASS)
+	ORDER by objid, refobjid;
+INSERT INTO cmtest VALUES(repeat('1234567890',1001));
+-- one normal dependency
+SELECT * FROM cmtest_deps;
+ classid | objsubid | refclassid | refobjsubid | deptype 
+---------+----------+------------+-------------+---------
+    1259 |        1 |       5555 |           0 | n
+(1 row)
+
+-- check decompression
+SELECT length(f1) FROM cmtest;
+ length 
+--------
+  10010
+(1 row)
+
+SELECT length(f1) FROM cmtest;
+ length 
+--------
+  10010
+(1 row)
+
+CREATE FUNCTION on_cmtest_rewrite()
+RETURNS event_trigger AS $$
+BEGIN
+	RAISE NOTICE 'cmtest rewrite';
+END;
+$$  LANGUAGE plpgsql;
+CREATE EVENT TRIGGER notice_on_cmtest_rewrite ON table_rewrite
+	EXECUTE PROCEDURE on_cmtest_rewrite();
+-- no rewrite
+ALTER TABLE cmtest ALTER COLUMN f1 SET COMPRESSION pglz1 PRESERVE (pglz);
+INSERT INTO cmtest VALUES(repeat('1234567890',1002));
+SELECT length(f1) FROM cmtest;
+ length 
+--------
+  10010
+  10020
+(2 rows)
+
+SELECT pg_column_compression('cmtest', 'f1');
+ pg_column_compression 
+-----------------------
+ pglz, pglz1
+(1 row)
+
+-- one normal and one internal dependency
+SELECT * FROM cmtest_deps;
+ classid | objsubid | refclassid | refobjsubid | deptype 
+---------+----------+------------+-------------+---------
+    1259 |        1 |       5555 |           0 | n
+    5555 |        0 |       1259 |           1 | i
+(2 rows)
+
+-- rewrite
+ALTER TABLE cmtest ALTER COLUMN f1 SET COMPRESSION pglz2 PRESERVE (pglz1);
+NOTICE:  cmtest rewrite
+INSERT INTO cmtest VALUES(repeat('1234567890',1003));
+SELECT length(f1) FROM cmtest;
+ length 
+--------
+  10010
+  10020
+  10030
+(3 rows)
+
+SELECT pg_column_compression('cmtest', 'f1');
+ pg_column_compression 
+-----------------------
+ pglz1, pglz2
+(1 row)
+
+-- two internal dependencies
+SELECT * FROM cmtest_deps;
+ classid | objsubid | refclassid | refobjsubid | deptype 
+---------+----------+------------+-------------+---------
+    5555 |        0 |       1259 |           1 | i
+    5555 |        0 |       1259 |           1 | i
+(2 rows)
+
+-- rewrite
+ALTER TABLE cmtest ALTER COLUMN f1 SET COMPRESSION pglz;
+NOTICE:  cmtest rewrite
+INSERT INTO cmtest VALUES(repeat('1234567890',1004));
+SELECT length(f1) FROM cmtest;
+ length 
+--------
+  10010
+  10020
+  10030
+  10040
+(4 rows)
+
+SELECT pg_column_compression('cmtest', 'f1');
+ pg_column_compression 
+-----------------------
+ pglz
+(1 row)
+
+-- one nornal dependency
+SELECT * FROM cmtest_deps;
+ classid | objsubid | refclassid | refobjsubid | deptype 
+---------+----------+------------+-------------+---------
+    1259 |        1 |       5555 |           0 | n
+(1 row)
+
+-- no rewrites
+ALTER TABLE cmtest ALTER COLUMN f1 SET COMPRESSION pglz1 PRESERVE (pglz);
+INSERT INTO cmtest VALUES(repeat('1234567890',1005));
+ALTER TABLE cmtest ALTER COLUMN f1 SET COMPRESSION pglz
+	WITH (min_input_size '1000') PRESERVE (pglz, pglz1);
+INSERT INTO cmtest VALUES(repeat('1234567890',1006));
+-- one nornal dependency and two internal dependencies
+SELECT * FROM cmtest_deps;
+ classid | objsubid | refclassid | refobjsubid | deptype 
+---------+----------+------------+-------------+---------
+    1259 |        1 |       5555 |           0 | n
+    5555 |        0 |       1259 |           1 | i
+    5555 |        0 |       1259 |           1 | i
+(3 rows)
+
+-- remove function and related event trigger
+DROP FUNCTION on_cmtest_rewrite CASCADE;
+NOTICE:  drop cascades to event trigger notice_on_cmtest_rewrite
+-- test moving
+CREATE TABLE cmdata(f1 text COMPRESSION pglz WITH (first_success_by '10000'));
 INSERT INTO cmdata VALUES(repeat('1234567890',1000));
 INSERT INTO cmdata VALUES(repeat('1234567890',1001));
 -- copy with table creation
 SELECT * INTO cmmove1 FROM cmdata;
 -- we update using datum from different table
-CREATE TABLE cmmove2(f1 text COMPRESSION pglz);
+CREATE TABLE cmmove2(f1 text COMPRESSION pglz WITH (min_input_size '100'));
 INSERT INTO cmmove2 VALUES (repeat('1234567890',1004));
 UPDATE cmmove2 SET f1 = cmdata.f1 FROM cmdata;
 -- copy to existing table
-CREATE TABLE cmmove3(f1 text COMPRESSION pglz);
+CREATE TABLE cmmove3(f1 text COMPRESSION pglz WITH (min_input_size '100'));
 INSERT INTO cmmove3 SELECT * FROM cmdata;
 -- drop original compression information
 DROP TABLE cmdata;
--- check data is okdd
+-- check data is ok
 SELECT length(f1) FROM cmmove1;
  length 
 --------
@@ -68,13 +288,108 @@ SELECT length(f1) FROM cmmove3;
   10010
 (2 rows)
 
+-- create different types of tables
+CREATE TABLE cmexample (f1 TEXT COMPRESSION pglz1);
+CREATE TABLE cmtestlike1 (LIKE cmexample INCLUDING COMPRESSION);
+CREATE TABLE cmtestlike2 (f2 INT) INHERITS (cmexample);
+\d+ cmtestlike1
+                                      Table "public.cmtestlike1"
+ Column | Type | Collation | Nullable | Default | Storage  | Compression | Stats target | Description 
+--------+------+-----------+----------+---------+----------+-------------+--------------+-------------
+ f1     | text |           |          |         | extended | pglz1       |              | 
+
+\d+ cmtestlike2
+                                       Table "public.cmtestlike2"
+ Column |  Type   | Collation | Nullable | Default | Storage  | Compression | Stats target | Description 
+--------+---------+-----------+----------+---------+----------+-------------+--------------+-------------
+ f1     | text    |           |          |         | extended | pglz1       |              | 
+ f2     | integer |           |          |         | plain    |             |              | 
+Inherits: cmexample
+
+-- test two columns
+CREATE TABLE cmaltertest(f1 TEXT, f2 TEXT COMPRESSION pglz1);
+\d+ cmaltertest;
+                                      Table "public.cmaltertest"
+ Column | Type | Collation | Nullable | Default | Storage  | Compression | Stats target | Description 
+--------+------+-----------+----------+---------+----------+-------------+--------------+-------------
+ f1     | text |           |          |         | extended | pglz        |              | 
+ f2     | text |           |          |         | extended | pglz1       |              | 
+
+-- fail, changing one column twice
+ALTER TABLE cmaltertest ALTER COLUMN f1 SET COMPRESSION pglz,
+	ALTER COLUMN f1 SET COMPRESSION pglz;
+ERROR:  cannot alter compression of column "f1" twice
+HINT:  Remove one of statements from the command.
+-- with rewrite
+ALTER TABLE cmaltertest ALTER COLUMN f1 SET COMPRESSION pglz1,
+	ALTER COLUMN f2 SET COMPRESSION pglz PRESERVE (pglz1);
+SELECT pg_column_compression('cmaltertest', 'f1');
+ pg_column_compression 
+-----------------------
+ pglz1
+(1 row)
+
+SELECT pg_column_compression('cmaltertest', 'f2');
+ pg_column_compression 
+-----------------------
+ pglz, pglz1
+(1 row)
+
+-- no rewrite
+ALTER TABLE cmaltertest ALTER COLUMN f1 SET COMPRESSION pglz PRESERVE (pglz1),
+	ALTER COLUMN f2 SET COMPRESSION pglz2 PRESERVE (pglz1, pglz);
+SELECT pg_column_compression('cmaltertest', 'f1');
+ pg_column_compression 
+-----------------------
+ pglz, pglz1
+(1 row)
+
+SELECT pg_column_compression('cmaltertest', 'f2');
+ pg_column_compression 
+-----------------------
+ pglz, pglz1, pglz2
+(1 row)
+
+-- make pglz2 droppable
+ALTER TABLE cmaltertest ALTER COLUMN f2 SET COMPRESSION pglz1;
+SELECT pg_column_compression('cmaltertest', 'f1');
+ pg_column_compression 
+-----------------------
+ pglz, pglz1
+(1 row)
+
+SELECT pg_column_compression('cmaltertest', 'f2');
+ pg_column_compression 
+-----------------------
+ pglz1
+(1 row)
+
+SELECT acname, acattnum, acoptions FROM pg_attr_compression
+	WHERE acrelid = 'cmaltertest'::REGCLASS OR acrelid = 'cmtest'::REGCLASS;
+ acname | acattnum |       acoptions       
+--------+----------+-----------------------
+ pglz1  |        1 | 
+ pglz   |        1 | {min_input_size=1000}
+ pglz1  |        2 | 
+ pglz1  |        1 | 
+(4 rows)
+
 -- zlib compression
+CREATE TABLE zlibtest(f1 TEXT COMPRESSION zlib WITH (invalid 'param'));
+ERROR:  unexpected parameter for zlib: "invalid"
+CREATE TABLE zlibtest(f1 TEXT COMPRESSION zlib WITH (level 'best'));
+ERROR:  invalid input syntax for type integer: "best"
 CREATE TABLE zlibtest(f1 TEXT COMPRESSION zlib);
-CREATE TABLE zlibtest(f1 TEXT COMPRESSION zlib);
-ERROR:  relation "zlibtest" already exists
-CREATE TABLE zlibtest(f1 TEXT COMPRESSION zlib);
-ERROR:  relation "zlibtest" already exists
+ALTER TABLE zlibtest
+	ALTER COLUMN f1 SET COMPRESSION zlib WITH (level '9');
+ALTER TABLE zlibtest
+	ALTER COLUMN f1 SET COMPRESSION zlib WITH (level '1');
+ALTER TABLE zlibtest
+	ALTER COLUMN f1 SET COMPRESSION zlib WITH (dict 'one');
+ERROR:  zlib dictionary is too small
 INSERT INTO zlibtest VALUES(repeat('1234567890',1004));
+ALTER TABLE zlibtest
+	ALTER COLUMN f1 SET COMPRESSION zlib WITH (dict 'one two') PRESERVE (zlib);
 INSERT INTO zlibtest VALUES(repeat('1234567890 one two three',1004));
 SELECT length(f1) FROM zlibtest;
  length 
@@ -83,4 +398,7 @@ SELECT length(f1) FROM zlibtest;
   24096
 (2 rows)
 
-DROP TABLE cmmove1, cmmove2, cmmove3, zlibtest;
+DROP ACCESS METHOD pglz2;
+DROP VIEW cmtest_deps;
+DROP TABLE cmmove1, cmmove2, cmmove3;
+DROP TABLE cmexample, cmtestlike1, cmtestlike2, zlibtest;
diff --git a/src/test/regress/expected/create_cm_1.out b/src/test/regress/expected/create_cm_1.out
new file mode 100644
index 0000000000..e38ae0cecf
--- /dev/null
+++ b/src/test/regress/expected/create_cm_1.out
@@ -0,0 +1,407 @@
+-- test drop
+DROP ACCESS METHOD pglz; --fail
+ERROR:  cannot drop access method pglz because it is required by the database system
+CREATE ACCESS METHOD pglz1 TYPE COMPRESSION HANDLER pglzhandler;
+CREATE TABLE droptest(d1 TEXT COMPRESSION pglz1);
+DROP ACCESS METHOD pglz1;
+ERROR:  cannot drop access method pglz1 because other objects depend on it
+DETAIL:  column d1 of table droptest depends on access method pglz1
+HINT:  Use DROP ... CASCADE to drop the dependent objects too.
+DROP ACCESS METHOD pglz1 CASCADE;
+NOTICE:  drop cascades to column d1 of table droptest
+\d+ droptest
+                                       Table "public.droptest"
+ Column | Type | Collation | Nullable | Default | Storage | Compression | Stats target | Description 
+--------+------+-----------+----------+---------+---------+-------------+--------------+-------------
+
+DROP TABLE droptest;
+CREATE ACCESS METHOD pglz1 TYPE COMPRESSION HANDLER pglzhandler;
+-- test auto drop of related attribute compression
+CREATE ACCESS METHOD pglz2 TYPE COMPRESSION HANDLER pglzhandler;
+CREATE TABLE cmaltertest (f1 TEXT COMPRESSION pglz2);
+SELECT acname, acattnum, acoptions FROM pg_attr_compression
+	WHERE acrelid = 'cmaltertest'::REGCLASS;
+ acname | acattnum | acoptions 
+--------+----------+-----------
+ pglz2  |        1 | 
+(1 row)
+
+ALTER TABLE cmaltertest DROP COLUMN f1;
+SELECT acname, acattnum, acoptions FROM pg_attr_compression
+	WHERE acrelid = 'cmaltertest'::REGCLASS;
+ acname | acattnum | acoptions 
+--------+----------+-----------
+(0 rows)
+
+DROP TABLE cmaltertest;
+-- test drop
+DROP ACCESS METHOD pglz2;
+-- test alter data type
+CREATE TABLE cmaltertest(at1 TEXT);
+ALTER TABLE cmaltertest ALTER COLUMN at1 SET COMPRESSION pglz1 PRESERVE (pglz);
+SELECT pg_column_compression('cmaltertest', 'at1');
+ pg_column_compression 
+-----------------------
+ pglz, pglz1
+(1 row)
+
+ALTER TABLE cmaltertest ALTER COLUMN at1 SET DATA TYPE INT USING at1::INTEGER;
+\d+ cmaltertest
+                                       Table "public.cmaltertest"
+ Column |  Type   | Collation | Nullable | Default | Storage | Compression | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+-------------+--------------+-------------
+ at1    | integer |           |          |         | plain   |             |              | 
+
+SELECT pg_column_compression('cmaltertest', 'at1');
+ pg_column_compression 
+-----------------------
+ 
+(1 row)
+
+SELECT acname, acattnum, acoptions FROM pg_attr_compression
+	WHERE acrelid = 'cmaltertest'::REGCLASS;
+ acname | acattnum | acoptions 
+--------+----------+-----------
+(0 rows)
+
+ALTER TABLE cmaltertest ALTER COLUMN at1 SET DATA TYPE TEXT;
+SELECT pg_column_compression('cmaltertest', 'at1');
+ pg_column_compression 
+-----------------------
+ pglz
+(1 row)
+
+DROP TABLE cmaltertest;
+-- test storages
+CREATE TABLE cmstoragetest(st1 TEXT, st2 INT);
+ALTER TABLE cmstoragetest ALTER COLUMN st2
+	SET COMPRESSION pglz WITH (min_input_size '100'); -- fail
+ERROR:  column data type integer does not support compression
+ALTER TABLE cmstoragetest ALTER COLUMN st1
+	SET COMPRESSION pglz WITH (min_input_size '100', min_comp_rate '50');
+ALTER TABLE cmstoragetest ALTER COLUMN st1 SET STORAGE EXTERNAL;
+\d+ cmstoragetest
+                                                        Table "public.cmstoragetest"
+ Column |  Type   | Collation | Nullable | Default | Storage  |                  Compression                   | Stats target | Description 
+--------+---------+-----------+----------+---------+----------+------------------------------------------------+--------------+-------------
+ st1    | text    |           |          |         | external | pglz(min_comp_rate '50', min_input_size '100') |              | 
+ st2    | integer |           |          |         | plain    |                                                |              | 
+
+ALTER TABLE cmstoragetest ALTER COLUMN st1 SET STORAGE MAIN;
+\d+ cmstoragetest
+                                                       Table "public.cmstoragetest"
+ Column |  Type   | Collation | Nullable | Default | Storage |                  Compression                   | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+------------------------------------------------+--------------+-------------
+ st1    | text    |           |          |         | main    | pglz(min_comp_rate '50', min_input_size '100') |              | 
+ st2    | integer |           |          |         | plain   |                                                |              | 
+
+ALTER TABLE cmstoragetest ALTER COLUMN st1 SET STORAGE PLAIN;
+\d+ cmstoragetest
+                                                       Table "public.cmstoragetest"
+ Column |  Type   | Collation | Nullable | Default | Storage |                  Compression                   | Stats target | Description 
+--------+---------+-----------+----------+---------+---------+------------------------------------------------+--------------+-------------
+ st1    | text    |           |          |         | plain   | pglz(min_comp_rate '50', min_input_size '100') |              | 
+ st2    | integer |           |          |         | plain   |                                                |              | 
+
+ALTER TABLE cmstoragetest ALTER COLUMN st1 SET STORAGE EXTENDED;
+\d+ cmstoragetest
+                                                        Table "public.cmstoragetest"
+ Column |  Type   | Collation | Nullable | Default | Storage  |                  Compression                   | Stats target | Description 
+--------+---------+-----------+----------+---------+----------+------------------------------------------------+--------------+-------------
+ st1    | text    |           |          |         | extended | pglz(min_comp_rate '50', min_input_size '100') |              | 
+ st2    | integer |           |          |         | plain    |                                                |              | 
+
+DROP TABLE cmstoragetest;
+CREATE ACCESS METHOD pglz2 TYPE COMPRESSION HANDLER pglzhandler;
+-- test PRESERVE
+CREATE TABLE cmtest(f1 TEXT);
+\d+ cmtest
+                                        Table "public.cmtest"
+ Column | Type | Collation | Nullable | Default | Storage  | Compression | Stats target | Description 
+--------+------+-----------+----------+---------+----------+-------------+--------------+-------------
+ f1     | text |           |          |         | extended | pglz        |              | 
+
+-- view to check dependencies
+CREATE VIEW cmtest_deps AS
+	SELECT classid, objsubid, refclassid, refobjsubid, deptype
+	FROM pg_depend
+	WHERE (refclassid = 'pg_catalog.pg_attr_compression'::REGCLASS OR
+            classid = 'pg_catalog.pg_attr_compression'::REGCLASS) AND
+		  (objid = 'cmtest'::REGCLASS OR refobjid = 'cmtest'::REGCLASS)
+	ORDER by objid, refobjid;
+INSERT INTO cmtest VALUES(repeat('1234567890',1001));
+-- one normal dependency
+SELECT * FROM cmtest_deps;
+ classid | objsubid | refclassid | refobjsubid | deptype 
+---------+----------+------------+-------------+---------
+    1259 |        1 |       5555 |           0 | n
+(1 row)
+
+-- check decompression
+SELECT length(f1) FROM cmtest;
+ length 
+--------
+  10010
+(1 row)
+
+SELECT length(f1) FROM cmtest;
+ length 
+--------
+  10010
+(1 row)
+
+CREATE FUNCTION on_cmtest_rewrite()
+RETURNS event_trigger AS $$
+BEGIN
+	RAISE NOTICE 'cmtest rewrite';
+END;
+$$  LANGUAGE plpgsql;
+CREATE EVENT TRIGGER notice_on_cmtest_rewrite ON table_rewrite
+	EXECUTE PROCEDURE on_cmtest_rewrite();
+-- no rewrite
+ALTER TABLE cmtest ALTER COLUMN f1 SET COMPRESSION pglz1 PRESERVE (pglz);
+INSERT INTO cmtest VALUES(repeat('1234567890',1002));
+SELECT length(f1) FROM cmtest;
+ length 
+--------
+  10010
+  10020
+(2 rows)
+
+SELECT pg_column_compression('cmtest', 'f1');
+ pg_column_compression 
+-----------------------
+ pglz, pglz1
+(1 row)
+
+-- one normal and one internal dependency
+SELECT * FROM cmtest_deps;
+ classid | objsubid | refclassid | refobjsubid | deptype 
+---------+----------+------------+-------------+---------
+    1259 |        1 |       5555 |           0 | n
+    5555 |        0 |       1259 |           1 | i
+(2 rows)
+
+-- rewrite
+ALTER TABLE cmtest ALTER COLUMN f1 SET COMPRESSION pglz2 PRESERVE (pglz1);
+NOTICE:  cmtest rewrite
+INSERT INTO cmtest VALUES(repeat('1234567890',1003));
+SELECT length(f1) FROM cmtest;
+ length 
+--------
+  10010
+  10020
+  10030
+(3 rows)
+
+SELECT pg_column_compression('cmtest', 'f1');
+ pg_column_compression 
+-----------------------
+ pglz1, pglz2
+(1 row)
+
+-- two internal dependencies
+SELECT * FROM cmtest_deps;
+ classid | objsubid | refclassid | refobjsubid | deptype 
+---------+----------+------------+-------------+---------
+    5555 |        0 |       1259 |           1 | i
+    5555 |        0 |       1259 |           1 | i
+(2 rows)
+
+-- rewrite
+ALTER TABLE cmtest ALTER COLUMN f1 SET COMPRESSION pglz;
+NOTICE:  cmtest rewrite
+INSERT INTO cmtest VALUES(repeat('1234567890',1004));
+SELECT length(f1) FROM cmtest;
+ length 
+--------
+  10010
+  10020
+  10030
+  10040
+(4 rows)
+
+SELECT pg_column_compression('cmtest', 'f1');
+ pg_column_compression 
+-----------------------
+ pglz
+(1 row)
+
+-- one nornal dependency
+SELECT * FROM cmtest_deps;
+ classid | objsubid | refclassid | refobjsubid | deptype 
+---------+----------+------------+-------------+---------
+    1259 |        1 |       5555 |           0 | n
+(1 row)
+
+-- no rewrites
+ALTER TABLE cmtest ALTER COLUMN f1 SET COMPRESSION pglz1 PRESERVE (pglz);
+INSERT INTO cmtest VALUES(repeat('1234567890',1005));
+ALTER TABLE cmtest ALTER COLUMN f1 SET COMPRESSION pglz
+	WITH (min_input_size '1000') PRESERVE (pglz, pglz1);
+INSERT INTO cmtest VALUES(repeat('1234567890',1006));
+-- one nornal dependency and two internal dependencies
+SELECT * FROM cmtest_deps;
+ classid | objsubid | refclassid | refobjsubid | deptype 
+---------+----------+------------+-------------+---------
+    1259 |        1 |       5555 |           0 | n
+    5555 |        0 |       1259 |           1 | i
+    5555 |        0 |       1259 |           1 | i
+(3 rows)
+
+-- remove function and related event trigger
+DROP FUNCTION on_cmtest_rewrite CASCADE;
+NOTICE:  drop cascades to event trigger notice_on_cmtest_rewrite
+-- test moving
+CREATE TABLE cmdata(f1 text COMPRESSION pglz WITH (first_success_by '10000'));
+INSERT INTO cmdata VALUES(repeat('1234567890',1000));
+INSERT INTO cmdata VALUES(repeat('1234567890',1001));
+-- copy with table creation
+SELECT * INTO cmmove1 FROM cmdata;
+-- we update using datum from different table
+CREATE TABLE cmmove2(f1 text COMPRESSION pglz WITH (min_input_size '100'));
+INSERT INTO cmmove2 VALUES (repeat('1234567890',1004));
+UPDATE cmmove2 SET f1 = cmdata.f1 FROM cmdata;
+-- copy to existing table
+CREATE TABLE cmmove3(f1 text COMPRESSION pglz WITH (min_input_size '100'));
+INSERT INTO cmmove3 SELECT * FROM cmdata;
+-- drop original compression information
+DROP TABLE cmdata;
+-- check data is ok
+SELECT length(f1) FROM cmmove1;
+ length 
+--------
+  10000
+  10010
+(2 rows)
+
+SELECT length(f1) FROM cmmove2;
+ length 
+--------
+  10000
+(1 row)
+
+SELECT length(f1) FROM cmmove3;
+ length 
+--------
+  10000
+  10010
+(2 rows)
+
+-- create different types of tables
+CREATE TABLE cmexample (f1 TEXT COMPRESSION pglz1);
+CREATE TABLE cmtestlike1 (LIKE cmexample INCLUDING COMPRESSION);
+CREATE TABLE cmtestlike2 (f2 INT) INHERITS (cmexample);
+\d+ cmtestlike1
+                                      Table "public.cmtestlike1"
+ Column | Type | Collation | Nullable | Default | Storage  | Compression | Stats target | Description 
+--------+------+-----------+----------+---------+----------+-------------+--------------+-------------
+ f1     | text |           |          |         | extended | pglz1       |              | 
+
+\d+ cmtestlike2
+                                       Table "public.cmtestlike2"
+ Column |  Type   | Collation | Nullable | Default | Storage  | Compression | Stats target | Description 
+--------+---------+-----------+----------+---------+----------+-------------+--------------+-------------
+ f1     | text    |           |          |         | extended | pglz1       |              | 
+ f2     | integer |           |          |         | plain    |             |              | 
+Inherits: cmexample
+
+-- test two columns
+CREATE TABLE cmaltertest(f1 TEXT, f2 TEXT COMPRESSION pglz1);
+\d+ cmaltertest;
+                                      Table "public.cmaltertest"
+ Column | Type | Collation | Nullable | Default | Storage  | Compression | Stats target | Description 
+--------+------+-----------+----------+---------+----------+-------------+--------------+-------------
+ f1     | text |           |          |         | extended | pglz        |              | 
+ f2     | text |           |          |         | extended | pglz1       |              | 
+
+-- fail, changing one column twice
+ALTER TABLE cmaltertest ALTER COLUMN f1 SET COMPRESSION pglz,
+	ALTER COLUMN f1 SET COMPRESSION pglz;
+ERROR:  cannot alter compression of column "f1" twice
+HINT:  Remove one of statements from the command.
+-- with rewrite
+ALTER TABLE cmaltertest ALTER COLUMN f1 SET COMPRESSION pglz1,
+	ALTER COLUMN f2 SET COMPRESSION pglz PRESERVE (pglz1);
+SELECT pg_column_compression('cmaltertest', 'f1');
+ pg_column_compression 
+-----------------------
+ pglz1
+(1 row)
+
+SELECT pg_column_compression('cmaltertest', 'f2');
+ pg_column_compression 
+-----------------------
+ pglz, pglz1
+(1 row)
+
+-- no rewrite
+ALTER TABLE cmaltertest ALTER COLUMN f1 SET COMPRESSION pglz PRESERVE (pglz1),
+	ALTER COLUMN f2 SET COMPRESSION pglz2 PRESERVE (pglz1, pglz);
+SELECT pg_column_compression('cmaltertest', 'f1');
+ pg_column_compression 
+-----------------------
+ pglz, pglz1
+(1 row)
+
+SELECT pg_column_compression('cmaltertest', 'f2');
+ pg_column_compression 
+-----------------------
+ pglz, pglz1, pglz2
+(1 row)
+
+-- make pglz2 droppable
+ALTER TABLE cmaltertest ALTER COLUMN f2 SET COMPRESSION pglz1;
+SELECT pg_column_compression('cmaltertest', 'f1');
+ pg_column_compression 
+-----------------------
+ pglz, pglz1
+(1 row)
+
+SELECT pg_column_compression('cmaltertest', 'f2');
+ pg_column_compression 
+-----------------------
+ pglz1
+(1 row)
+
+SELECT acname, acattnum, acoptions FROM pg_attr_compression
+	WHERE acrelid = 'cmaltertest'::REGCLASS OR acrelid = 'cmtest'::REGCLASS;
+ acname | acattnum |       acoptions       
+--------+----------+-----------------------
+ pglz1  |        1 | 
+ pglz   |        1 | {min_input_size=1000}
+ pglz1  |        2 | 
+ pglz1  |        1 | 
+(4 rows)
+
+-- zlib compression
+CREATE TABLE zlibtest(f1 TEXT COMPRESSION zlib WITH (invalid 'param'));
+ERROR:  not built with zlib support
+CREATE TABLE zlibtest(f1 TEXT COMPRESSION zlib WITH (level 'best'));
+ERROR:  not built with zlib support
+CREATE TABLE zlibtest(f1 TEXT COMPRESSION zlib);
+ALTER TABLE zlibtest
+	ALTER COLUMN f1 SET COMPRESSION zlib WITH (level '9');
+ERROR:  not built with zlib support
+ALTER TABLE zlibtest
+	ALTER COLUMN f1 SET COMPRESSION zlib WITH (level '1');
+ERROR:  not built with zlib support
+ALTER TABLE zlibtest
+	ALTER COLUMN f1 SET COMPRESSION zlib WITH (dict 'one');
+ERROR:  not built with zlib support
+INSERT INTO zlibtest VALUES(repeat('1234567890',1004));
+ERROR:  not built with zlib support
+ALTER TABLE zlibtest
+	ALTER COLUMN f1 SET COMPRESSION zlib WITH (dict 'one two') PRESERVE (zlib);
+ERROR:  not built with zlib support
+INSERT INTO zlibtest VALUES(repeat('1234567890 one two three',1004));
+ERROR:  not built with zlib support
+SELECT length(f1) FROM zlibtest;
+ length 
+--------
+(0 rows)
+
+DROP ACCESS METHOD pglz2;
+DROP VIEW cmtest_deps;
+DROP TABLE cmmove1, cmmove2, cmmove3;
+DROP TABLE cmexample, cmtestlike1, cmtestlike2, zlibtest;
diff --git a/src/test/regress/expected/create_index.out b/src/test/regress/expected/create_index.out
index e3e6634d7e..2f3bc66435 100644
--- a/src/test/regress/expected/create_index.out
+++ b/src/test/regress/expected/create_index.out
@@ -2064,6 +2064,7 @@ WHERE classid = 'pg_class'::regclass AND
   ORDER BY 1, 2;
                    obj                    |                           objref                           | deptype 
 ------------------------------------------+------------------------------------------------------------+---------
+ column c2 of table concur_reindex_tab    | attribute compression 4225                                 | n
  index concur_reindex_ind1                | constraint concur_reindex_ind1 on table concur_reindex_tab | i
  index concur_reindex_ind2                | column c2 of table concur_reindex_tab                      | a
  index concur_reindex_ind3                | column c1 of table concur_reindex_tab                      | a
@@ -2073,7 +2074,7 @@ WHERE classid = 'pg_class'::regclass AND
  index concur_reindex_ind4                | column c2 of table concur_reindex_tab                      | a
  materialized view concur_reindex_matview | schema public                                              | n
  table concur_reindex_tab                 | schema public                                              | n
-(9 rows)
+(10 rows)
 
 REINDEX INDEX CONCURRENTLY concur_reindex_ind1;
 REINDEX TABLE CONCURRENTLY concur_reindex_tab;
@@ -2092,6 +2093,7 @@ WHERE classid = 'pg_class'::regclass AND
   ORDER BY 1, 2;
                    obj                    |                           objref                           | deptype 
 ------------------------------------------+------------------------------------------------------------+---------
+ column c2 of table concur_reindex_tab    | attribute compression 4225                                 | n
  index concur_reindex_ind1                | constraint concur_reindex_ind1 on table concur_reindex_tab | i
  index concur_reindex_ind2                | column c2 of table concur_reindex_tab                      | a
  index concur_reindex_ind3                | column c1 of table concur_reindex_tab                      | a
@@ -2101,7 +2103,7 @@ WHERE classid = 'pg_class'::regclass AND
  index concur_reindex_ind4                | column c2 of table concur_reindex_tab                      | a
  materialized view concur_reindex_matview | schema public                                              | n
  table concur_reindex_tab                 | schema public                                              | n
-(9 rows)
+(10 rows)
 
 -- Check that comments are preserved
 CREATE TABLE testcomment (i int);
diff --git a/src/test/regress/expected/psql.out b/src/test/regress/expected/psql.out
index 531b298fe3..6fab6cddd8 100644
--- a/src/test/regress/expected/psql.out
+++ b/src/test/regress/expected/psql.out
@@ -4844,9 +4844,10 @@ List of access methods
  heap   | Table
  heap2  | Table
  pglz   | 
+ pglz1  | 
  spgist | Index
  zlib   | 
-(10 rows)
+(11 rows)
 
 \dA *
 List of access methods
@@ -4860,9 +4861,10 @@ List of access methods
  heap   | Table
  heap2  | Table
  pglz   | 
+ pglz1  | 
  spgist | Index
  zlib   | 
-(10 rows)
+(11 rows)
 
 \dA h*
 List of access methods
@@ -4898,9 +4900,10 @@ List of access methods
  heap   | Table | heap_tableam_handler | heap table access method
  heap2  | Table | heap_tableam_handler | 
  pglz   |       | pglzhandler          | pglz compression access method
+ pglz1  |       | pglzhandler          | 
  spgist | Index | spghandler           | SP-GiST index access method
  zlib   |       | zlibhandler          | zlib compression access method
-(10 rows)
+(11 rows)
 
 \dA+ *
                              List of access methods
@@ -4914,9 +4917,10 @@ List of access methods
  heap   | Table | heap_tableam_handler | heap table access method
  heap2  | Table | heap_tableam_handler | 
  pglz   |       | pglzhandler          | pglz compression access method
+ pglz1  |       | pglzhandler          | 
  spgist | Index | spghandler           | SP-GiST index access method
  zlib   |       | zlibhandler          | zlib compression access method
-(10 rows)
+(11 rows)
 
 \dA+ h*
                      List of access methods
diff --git a/src/test/regress/expected/sanity_check.out b/src/test/regress/expected/sanity_check.out
index 192445878d..6b7f772c95 100644
--- a/src/test/regress/expected/sanity_check.out
+++ b/src/test/regress/expected/sanity_check.out
@@ -32,6 +32,8 @@ check2_tbl|f
 check_tbl|f
 circle_tbl|t
 city|f
+cmaltertest|f
+cmtest|f
 copy_tbl|f
 d|f
 d_star|f
@@ -104,6 +106,7 @@ pg_aggregate|t
 pg_am|t
 pg_amop|t
 pg_amproc|t
+pg_attr_compression|t
 pg_attrdef|t
 pg_attribute|t
 pg_auth_members|t
diff --git a/src/test/regress/sql/create_cm.sql b/src/test/regress/sql/create_cm.sql
index 56501b45b0..a598484253 100644
--- a/src/test/regress/sql/create_cm.sql
+++ b/src/test/regress/sql/create_cm.sql
@@ -1,5 +1,47 @@
+-- test drop
+DROP ACCESS METHOD pglz; --fail
+
+CREATE ACCESS METHOD pglz1 TYPE COMPRESSION HANDLER pglzhandler;
+CREATE TABLE droptest(d1 TEXT COMPRESSION pglz1);
+DROP ACCESS METHOD pglz1;
+DROP ACCESS METHOD pglz1 CASCADE;
+\d+ droptest
+DROP TABLE droptest;
+
+CREATE ACCESS METHOD pglz1 TYPE COMPRESSION HANDLER pglzhandler;
+
+-- test auto drop of related attribute compression
+CREATE ACCESS METHOD pglz2 TYPE COMPRESSION HANDLER pglzhandler;
+CREATE TABLE cmaltertest (f1 TEXT COMPRESSION pglz2);
+SELECT acname, acattnum, acoptions FROM pg_attr_compression
+	WHERE acrelid = 'cmaltertest'::REGCLASS;
+ALTER TABLE cmaltertest DROP COLUMN f1;
+SELECT acname, acattnum, acoptions FROM pg_attr_compression
+	WHERE acrelid = 'cmaltertest'::REGCLASS;
+DROP TABLE cmaltertest;
+
+-- test drop
+DROP ACCESS METHOD pglz2;
+
+-- test alter data type
+CREATE TABLE cmaltertest(at1 TEXT);
+ALTER TABLE cmaltertest ALTER COLUMN at1 SET COMPRESSION pglz1 PRESERVE (pglz);
+SELECT pg_column_compression('cmaltertest', 'at1');
+ALTER TABLE cmaltertest ALTER COLUMN at1 SET DATA TYPE INT USING at1::INTEGER;
+\d+ cmaltertest
+SELECT pg_column_compression('cmaltertest', 'at1');
+SELECT acname, acattnum, acoptions FROM pg_attr_compression
+	WHERE acrelid = 'cmaltertest'::REGCLASS;
+ALTER TABLE cmaltertest ALTER COLUMN at1 SET DATA TYPE TEXT;
+SELECT pg_column_compression('cmaltertest', 'at1');
+DROP TABLE cmaltertest;
+
 -- test storages
 CREATE TABLE cmstoragetest(st1 TEXT, st2 INT);
+ALTER TABLE cmstoragetest ALTER COLUMN st2
+	SET COMPRESSION pglz WITH (min_input_size '100'); -- fail
+ALTER TABLE cmstoragetest ALTER COLUMN st1
+	SET COMPRESSION pglz WITH (min_input_size '100', min_comp_rate '50');
 ALTER TABLE cmstoragetest ALTER COLUMN st1 SET STORAGE EXTERNAL;
 \d+ cmstoragetest
 ALTER TABLE cmstoragetest ALTER COLUMN st1 SET STORAGE MAIN;
@@ -10,7 +52,76 @@ ALTER TABLE cmstoragetest ALTER COLUMN st1 SET STORAGE EXTENDED;
 \d+ cmstoragetest
 DROP TABLE cmstoragetest;
 
-CREATE TABLE cmdata(f1 text COMPRESSION pglz);
+CREATE ACCESS METHOD pglz2 TYPE COMPRESSION HANDLER pglzhandler;
+
+-- test PRESERVE
+CREATE TABLE cmtest(f1 TEXT);
+\d+ cmtest
+-- view to check dependencies
+CREATE VIEW cmtest_deps AS
+	SELECT classid, objsubid, refclassid, refobjsubid, deptype
+	FROM pg_depend
+	WHERE (refclassid = 'pg_catalog.pg_attr_compression'::REGCLASS OR
+            classid = 'pg_catalog.pg_attr_compression'::REGCLASS) AND
+		  (objid = 'cmtest'::REGCLASS OR refobjid = 'cmtest'::REGCLASS)
+	ORDER by objid, refobjid;
+INSERT INTO cmtest VALUES(repeat('1234567890',1001));
+
+-- one normal dependency
+SELECT * FROM cmtest_deps;
+
+-- check decompression
+SELECT length(f1) FROM cmtest;
+SELECT length(f1) FROM cmtest;
+
+CREATE FUNCTION on_cmtest_rewrite()
+RETURNS event_trigger AS $$
+BEGIN
+	RAISE NOTICE 'cmtest rewrite';
+END;
+$$  LANGUAGE plpgsql;
+
+CREATE EVENT TRIGGER notice_on_cmtest_rewrite ON table_rewrite
+	EXECUTE PROCEDURE on_cmtest_rewrite();
+
+-- no rewrite
+ALTER TABLE cmtest ALTER COLUMN f1 SET COMPRESSION pglz1 PRESERVE (pglz);
+INSERT INTO cmtest VALUES(repeat('1234567890',1002));
+SELECT length(f1) FROM cmtest;
+SELECT pg_column_compression('cmtest', 'f1');
+-- one normal and one internal dependency
+SELECT * FROM cmtest_deps;
+
+-- rewrite
+ALTER TABLE cmtest ALTER COLUMN f1 SET COMPRESSION pglz2 PRESERVE (pglz1);
+INSERT INTO cmtest VALUES(repeat('1234567890',1003));
+SELECT length(f1) FROM cmtest;
+SELECT pg_column_compression('cmtest', 'f1');
+-- two internal dependencies
+SELECT * FROM cmtest_deps;
+
+-- rewrite
+ALTER TABLE cmtest ALTER COLUMN f1 SET COMPRESSION pglz;
+INSERT INTO cmtest VALUES(repeat('1234567890',1004));
+SELECT length(f1) FROM cmtest;
+SELECT pg_column_compression('cmtest', 'f1');
+-- one nornal dependency
+SELECT * FROM cmtest_deps;
+
+-- no rewrites
+ALTER TABLE cmtest ALTER COLUMN f1 SET COMPRESSION pglz1 PRESERVE (pglz);
+INSERT INTO cmtest VALUES(repeat('1234567890',1005));
+ALTER TABLE cmtest ALTER COLUMN f1 SET COMPRESSION pglz
+	WITH (min_input_size '1000') PRESERVE (pglz, pglz1);
+INSERT INTO cmtest VALUES(repeat('1234567890',1006));
+-- one nornal dependency and two internal dependencies
+SELECT * FROM cmtest_deps;
+
+-- remove function and related event trigger
+DROP FUNCTION on_cmtest_rewrite CASCADE;
+
+-- test moving
+CREATE TABLE cmdata(f1 text COMPRESSION pglz WITH (first_success_by '10000'));
 INSERT INTO cmdata VALUES(repeat('1234567890',1000));
 INSERT INTO cmdata VALUES(repeat('1234567890',1001));
 
@@ -18,28 +129,74 @@ INSERT INTO cmdata VALUES(repeat('1234567890',1001));
 SELECT * INTO cmmove1 FROM cmdata;
 
 -- we update using datum from different table
-CREATE TABLE cmmove2(f1 text COMPRESSION pglz);
+CREATE TABLE cmmove2(f1 text COMPRESSION pglz WITH (min_input_size '100'));
 INSERT INTO cmmove2 VALUES (repeat('1234567890',1004));
 UPDATE cmmove2 SET f1 = cmdata.f1 FROM cmdata;
 
 -- copy to existing table
-CREATE TABLE cmmove3(f1 text COMPRESSION pglz);
+CREATE TABLE cmmove3(f1 text COMPRESSION pglz WITH (min_input_size '100'));
 INSERT INTO cmmove3 SELECT * FROM cmdata;
 
 -- drop original compression information
 DROP TABLE cmdata;
 
--- check data is okdd
+-- check data is ok
 SELECT length(f1) FROM cmmove1;
 SELECT length(f1) FROM cmmove2;
 SELECT length(f1) FROM cmmove3;
 
+-- create different types of tables
+CREATE TABLE cmexample (f1 TEXT COMPRESSION pglz1);
+CREATE TABLE cmtestlike1 (LIKE cmexample INCLUDING COMPRESSION);
+CREATE TABLE cmtestlike2 (f2 INT) INHERITS (cmexample);
+
+\d+ cmtestlike1
+\d+ cmtestlike2
+
+-- test two columns
+CREATE TABLE cmaltertest(f1 TEXT, f2 TEXT COMPRESSION pglz1);
+\d+ cmaltertest;
+-- fail, changing one column twice
+ALTER TABLE cmaltertest ALTER COLUMN f1 SET COMPRESSION pglz,
+	ALTER COLUMN f1 SET COMPRESSION pglz;
+
+-- with rewrite
+ALTER TABLE cmaltertest ALTER COLUMN f1 SET COMPRESSION pglz1,
+	ALTER COLUMN f2 SET COMPRESSION pglz PRESERVE (pglz1);
+SELECT pg_column_compression('cmaltertest', 'f1');
+SELECT pg_column_compression('cmaltertest', 'f2');
+
+-- no rewrite
+ALTER TABLE cmaltertest ALTER COLUMN f1 SET COMPRESSION pglz PRESERVE (pglz1),
+	ALTER COLUMN f2 SET COMPRESSION pglz2 PRESERVE (pglz1, pglz);
+SELECT pg_column_compression('cmaltertest', 'f1');
+SELECT pg_column_compression('cmaltertest', 'f2');
+
+-- make pglz2 droppable
+ALTER TABLE cmaltertest ALTER COLUMN f2 SET COMPRESSION pglz1;
+SELECT pg_column_compression('cmaltertest', 'f1');
+SELECT pg_column_compression('cmaltertest', 'f2');
+
+SELECT acname, acattnum, acoptions FROM pg_attr_compression
+	WHERE acrelid = 'cmaltertest'::REGCLASS OR acrelid = 'cmtest'::REGCLASS;
+
 -- zlib compression
+CREATE TABLE zlibtest(f1 TEXT COMPRESSION zlib WITH (invalid 'param'));
+CREATE TABLE zlibtest(f1 TEXT COMPRESSION zlib WITH (level 'best'));
 CREATE TABLE zlibtest(f1 TEXT COMPRESSION zlib);
-CREATE TABLE zlibtest(f1 TEXT COMPRESSION zlib);
-CREATE TABLE zlibtest(f1 TEXT COMPRESSION zlib);
+ALTER TABLE zlibtest
+	ALTER COLUMN f1 SET COMPRESSION zlib WITH (level '9');
+ALTER TABLE zlibtest
+	ALTER COLUMN f1 SET COMPRESSION zlib WITH (level '1');
+ALTER TABLE zlibtest
+	ALTER COLUMN f1 SET COMPRESSION zlib WITH (dict 'one');
 INSERT INTO zlibtest VALUES(repeat('1234567890',1004));
+ALTER TABLE zlibtest
+	ALTER COLUMN f1 SET COMPRESSION zlib WITH (dict 'one two') PRESERVE (zlib);
 INSERT INTO zlibtest VALUES(repeat('1234567890 one two three',1004));
 SELECT length(f1) FROM zlibtest;
 
-DROP TABLE cmmove1, cmmove2, cmmove3, zlibtest;
+DROP ACCESS METHOD pglz2;
+DROP VIEW cmtest_deps;
+DROP TABLE cmmove1, cmmove2, cmmove3;
+DROP TABLE cmexample, cmtestlike1, cmtestlike2, zlibtest;
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 1953697af9..de06e3993f 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -367,6 +367,7 @@ CollectedCommand
 CollectedCommandType
 ColorTrgm
 ColorTrgmInfo
+ColumnCompression
 ColumnCompareData
 ColumnDef
 ColumnIOData
-- 
2.23.0