v15-0006-Support-compression-methods-options.patch

From ccf8862fcaed690752a14bc76288bd523dc7a2d9 Mon Sep 17 00:00:00 2001
From: dilipkumar <dilipbalaut@gmail.com>
Date: Tue, 8 Dec 2020 16:46:53 +0530
Subject: [PATCH v15 6/6] Support compression methods options

---
 contrib/cmzlib/cmzlib.c                       |  75 ++++++++--
 doc/src/sgml/ref/alter_table.sgml             |   6 +-
 doc/src/sgml/ref/create_table.sgml            |   8 +-
 src/backend/access/brin/brin_tuple.c          |   5 +-
 src/backend/access/common/indextuple.c        |   4 +-
 src/backend/access/common/reloptions.c        |  42 ++++++
 src/backend/access/common/toast_internals.c   |  14 +-
 src/backend/access/compression/compress_lz4.c |  69 ++++++++-
 .../access/compression/compress_pglz.c        |  98 ++++++++++++-
 src/backend/access/table/toast_helper.c       |   8 +-
 src/backend/bootstrap/bootparse.y             |   1 +
 src/backend/catalog/heap.c                    |  15 +-
 src/backend/catalog/index.c                   |  43 +++++-
 src/backend/catalog/toasting.c                |   1 +
 src/backend/commands/cluster.c                |   1 +
 src/backend/commands/compressioncmds.c        |  48 ++++++-
 src/backend/commands/foreigncmds.c            |  44 ------
 src/backend/commands/tablecmds.c              | 133 ++++++++++++++++--
 src/backend/nodes/copyfuncs.c                 |   1 +
 src/backend/nodes/equalfuncs.c                |   1 +
 src/backend/nodes/outfuncs.c                  |   1 +
 src/backend/parser/gram.y                     |  16 ++-
 src/include/access/compressamapi.h            |  19 ++-
 src/include/access/toast_helper.h             |   2 +
 src/include/access/toast_internals.h          |   2 +-
 src/include/catalog/heap.h                    |   2 +
 src/include/catalog/pg_attribute.h            |   3 +
 src/include/commands/defrem.h                 |   4 +-
 src/include/nodes/parsenodes.h                |   1 +
 src/test/regress/expected/compression.out     |  21 ++-
 src/test/regress/expected/misc_sanity.out     |   3 +-
 src/test/regress/sql/compression.sql          |  11 ++
 32 files changed, 594 insertions(+), 108 deletions(-)

diff --git a/contrib/cmzlib/cmzlib.c b/contrib/cmzlib/cmzlib.c
index 686a7c7e0d..cb5aa439e3 100644
--- a/contrib/cmzlib/cmzlib.c
+++ b/contrib/cmzlib/cmzlib.c
@@ -14,6 +14,7 @@
 #include "postgres.h"
 #include "access/compressamapi.h"
 #include "access/toast_internals.h"
+#include "commands/defrem.h"
 
 #include "fmgr.h"
 #include "utils/builtins.h"
@@ -45,6 +46,62 @@ 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(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
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_PARAMETER),
+					 errmsg("unexpected parameter for zlib: \"%s\"", def->defname)));
+	}
+}
+
+static void *
+zlib_cminitstate(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);
+		}
+	}
+
+	return state;
+}
+
 /*
  * zlib_cmcompress - compression routine for zlib compression method
  *
@@ -52,23 +109,21 @@ typedef struct
  * Returns the compressed varlena, or NULL if compression fails.
  */
 static struct varlena *
-zlib_cmcompress(const struct varlena *value, int32 header_size)
+zlib_cmcompress(const struct varlena *value, int32 header_size, void *options)
 {
-	int32		valsize,
-				len;
+	int32			valsize,
+					len;
 	struct varlena *tmp = NULL;
-	z_streamp	zp;
-	int			res;
-	zlib_state	state;
-
-	state.level = Z_DEFAULT_COMPRESSION;
+	z_streamp		zp;
+	int				res;
+	zlib_state	   *state = (zlib_state *) options;
 
 	zp = (z_streamp) palloc(sizeof(z_stream));
 	zp->zalloc = Z_NULL;
 	zp->zfree = Z_NULL;
 	zp->opaque = Z_NULL;
 
-	if (deflateInit(zp, state.level) != Z_OK)
+	if (deflateInit(zp, state->level) != Z_OK)
 		elog(ERROR, "could not initialize compression library: %s", zp->msg);
 
 	valsize = VARSIZE_ANY_EXHDR(DatumGetPointer(value));
@@ -146,6 +201,8 @@ zlib_cmdecompress(const struct varlena *value, int32 header_size)
 
 const CompressionAmRoutine zlib_compress_methods = {
 	.type = T_CompressionAmRoutine,
+	.datum_check = zlib_cmcheck,
+	.datum_initstate = zlib_cminitstate,
 	.datum_compress = zlib_cmcompress,
 	.datum_decompress = zlib_cmdecompress,
 	.datum_decompress_slice = NULL};
diff --git a/doc/src/sgml/ref/alter_table.sgml b/doc/src/sgml/ref/alter_table.sgml
index 9efbc2352c..87e1a1acf3 100644
--- a/doc/src/sgml/ref/alter_table.sgml
+++ b/doc/src/sgml/ref/alter_table.sgml
@@ -54,7 +54,7 @@ ALTER TABLE [ IF EXISTS ] <replaceable class="parameter">name</replaceable>
     ALTER [ COLUMN ] <replaceable class="parameter">column_name</replaceable> SET ( <replaceable class="parameter">attribute_option</replaceable> = <replaceable class="parameter">value</replaceable> [, ... ] )
     ALTER [ COLUMN ] <replaceable class="parameter">column_name</replaceable> RESET ( <replaceable class="parameter">attribute_option</replaceable> [, ... ] )
     ALTER [ COLUMN ] <replaceable class="parameter">column_name</replaceable> SET STORAGE { PLAIN | EXTERNAL | EXTENDED | MAIN }
-    ALTER [ COLUMN ] <replaceable class="parameter">column_name</replaceable> SET COMPRESSION <replaceable class="parameter">compression_method</replaceable> [ PRESERVE (<replaceable class="parameter">compression_preserve_list</replaceable>) | PRESERVE ALL ]
+    ALTER [ COLUMN ] <replaceable class="parameter">column_name</replaceable> SET COMPRESSION <replaceable class="parameter">compression_method</replaceable> [ WITH (<replaceable class="parameter">compression_method_options</replaceable>) ] [ PRESERVE (<replaceable class="parameter">compression_preserve_list</replaceable>) | PRESERVE ALL ]
     ADD <replaceable class="parameter">table_constraint</replaceable> [ NOT VALID ]
     ADD <replaceable class="parameter">table_constraint_using_index</replaceable>
     ALTER CONSTRAINT <replaceable class="parameter">constraint_name</replaceable> [ DEFERRABLE | NOT DEFERRABLE ] [ INITIALLY DEFERRED | INITIALLY IMMEDIATE ]
@@ -386,7 +386,7 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
 
    <varlistentry>
     <term>
-     <literal>SET COMPRESSION <replaceable class="parameter">compression_method</replaceable> [ PRESERVE (<replaceable class="parameter">compression_preserve_list</replaceable>) | PRESERVE ALL ]</literal>
+     <literal>SET COMPRESSION <replaceable class="parameter">compression_method</replaceable> [ WITH (<replaceable class="parameter">compression_method_options</replaceable>) ] [ PRESERVE (<replaceable class="parameter">compression_preserve_list</replaceable>) | PRESERVE ALL ]</literal>
     </term>
     <listitem>
      <para>
@@ -394,6 +394,8 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
       could be created with <xref linkend="sql-create-access-method"/> or it can
       be set from the available built-in compression methods.  The available
       built-in methods are <literal>pglz</literal> and <literal>lz4</literal>.
+      If compression method has options they could be specified with
+      <literal>WITH</literal> parameter.
       The PRESERVE list contains list of compression methods used on the column
       and determines which of them should be kept on the column.  Without
       PRESERVE or if all the previous compression methods are not preserved then
diff --git a/doc/src/sgml/ref/create_table.sgml b/doc/src/sgml/ref/create_table.sgml
index ade3989d75..95e03f6379 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_method</replaceable> |
+  COMPRESSION <replaceable class="parameter">compression_method</replaceable> [ WITH (<replaceable class="parameter">compression_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 ]
@@ -995,7 +995,7 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
    </varlistentry>
 
    <varlistentry>
-    <term><literal>COMPRESSION <replaceable class="parameter">compression_method</replaceable></literal></term>
+    <term><literal>COMPRESSION <replaceable class="parameter">compression_method</replaceable> [ WITH (<replaceable class="parameter">compression_am_options</replaceable>) ]</literal></term>
     <listitem>
      <para>
       This clause adds the compression method to a column.  Compression method
@@ -1004,7 +1004,9 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
       built-in methods are <literal>pglz</literal> and <literal>lz4</literal>.
       If the compression method is not sepcified for the compressible type then
       it will have the default compression method.  The default compression
-      method is <literal>pglz</literal>.
+      method is <literal>pglz</literal>.  If the compression method has options
+      they could be specified by <literal>WITH</literal>
+      parameter.
      </para>
     </listitem>
    </varlistentry>
diff --git a/src/backend/access/brin/brin_tuple.c b/src/backend/access/brin/brin_tuple.c
index 2b8a6de6ed..61fc6fbb30 100644
--- a/src/backend/access/brin/brin_tuple.c
+++ b/src/backend/access/brin/brin_tuple.c
@@ -38,6 +38,7 @@
 #include "access/toast_internals.h"
 #include "access/tupdesc.h"
 #include "access/tupmacs.h"
+#include "commands/defrem.h"
 #include "utils/datum.h"
 #include "utils/memutils.h"
 
@@ -215,8 +216,10 @@ brin_form_tuple(BrinDesc *brdesc, BlockNumber blkno, BrinMemTuple *tuple,
 			{
 				Form_pg_attribute att = TupleDescAttr(brdesc->bd_tupdesc,
 													  keyno);
+				List	   *acoption = GetAttributeCompressionOptions(att);
 				Datum		cvalue = toast_compress_datum(value,
-														  att->attcompression);
+														  att->attcompression,
+														  acoption);
 
 				if (DatumGetPointer(cvalue) != NULL)
 				{
diff --git a/src/backend/access/common/indextuple.c b/src/backend/access/common/indextuple.c
index 3024670414..0cd7bdd730 100644
--- a/src/backend/access/common/indextuple.c
+++ b/src/backend/access/common/indextuple.c
@@ -22,6 +22,7 @@
 #include "access/htup_details.h"
 #include "access/itup.h"
 #include "access/toast_internals.h"
+#include "commands/defrem.h"
 
 /*
  * This enables de-toasting of index entries.  Needed until VACUUM is
@@ -104,8 +105,9 @@ index_form_tuple(TupleDesc tupleDescriptor,
 			(att->attstorage == TYPSTORAGE_EXTENDED ||
 			 att->attstorage == TYPSTORAGE_MAIN))
 		{
+			List	   *acoption = GetAttributeCompressionOptions(att);
 			Datum		cvalue = toast_compress_datum(untoasted_values[i],
-													  att->attcompression);
+													  att->attcompression, acoption);
 
 			if (DatumGetPointer(cvalue) != NULL)
 			{
diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index 8ccc228a8c..3185082034 100644
--- a/src/backend/access/common/reloptions.c
+++ b/src/backend/access/common/reloptions.c
@@ -2112,3 +2112,45 @@ AlterTableGetRelOptionsLockLevel(List *defList)
 
 	return lockmode;
 }
+
+/*
+ * 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.
+ */
+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);
+}
diff --git a/src/backend/access/common/toast_internals.c b/src/backend/access/common/toast_internals.c
index cd91aefef7..e9b5971928 100644
--- a/src/backend/access/common/toast_internals.c
+++ b/src/backend/access/common/toast_internals.c
@@ -44,10 +44,11 @@ static bool toastid_valueid_exists(Oid toastrelid, Oid valueid);
  * ----------
  */
 Datum
-toast_compress_datum(Datum value, Oid cmoid)
+toast_compress_datum(Datum value, Oid cmoid, List *cmoptions)
 {
 	struct varlena *tmp = NULL;
 	int32		valsize;
+	void	   *options = NULL;
 	bool		isCustomCompression = false;
 	const CompressionAmRoutine *cmroutine = NULL;
 
@@ -71,11 +72,20 @@ toast_compress_datum(Datum value, Oid cmoid)
 			break;
 	}
 
+	if (cmroutine->datum_initstate)
+		options = cmroutine->datum_initstate(cmoptions);
+
 	/* Call the actual compression function */
 	tmp = cmroutine->datum_compress((const struct varlena *)value,
 									isCustomCompression ?
 									TOAST_CUSTOM_COMPRESS_HDRSZ :
-									TOAST_COMPRESS_HDRSZ);
+									TOAST_COMPRESS_HDRSZ, options);
+	if (options != NULL)
+	{
+		pfree(options);
+		list_free(cmoptions);
+	}
+
 	if (!tmp)
 		return PointerGetDatum(NULL);
 
diff --git a/src/backend/access/compression/compress_lz4.c b/src/backend/access/compression/compress_lz4.c
index ce27616116..d1ee63c8e9 100644
--- a/src/backend/access/compression/compress_lz4.c
+++ b/src/backend/access/compression/compress_lz4.c
@@ -16,12 +16,70 @@
 #include "postgres.h"
 #include "access/compressamapi.h"
 #include "access/toast_internals.h"
+#include "commands/defrem.h"
 #include "fmgr.h"
 #include "utils/builtins.h"
 
 #ifdef HAVE_LIBLZ4
 #include "lz4.h"
 
+/*
+ * Check options if specified. All validation is located here so
+ * we don't need do it again in cminitstate function.
+ */
+static void
+lz4_cmcheck(List *options)
+{
+	ListCell	*lc;
+
+	foreach(lc, options)
+	{
+		DefElem    *def = (DefElem *) lfirst(lc);
+
+		if (strcmp(def->defname, "acceleration") == 0)
+		{
+			int32 acceleration =
+				pg_atoi(defGetString(def), sizeof(acceleration), 0);
+
+			if (acceleration < INT32_MIN || acceleration > INT32_MAX)
+				ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("unexpected value for lz4 compression acceleration: \"%s\"",
+								defGetString(def)),
+					 errhint("expected value between INT32_MIN and INT32_MAX")
+					));
+		}
+		else
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_PARAMETER),
+					 errmsg("unexpected parameter for lz4: \"%s\"", def->defname)));
+	}
+}
+
+static void *
+lz4_cminitstate(List *options)
+{
+	int32	*acceleration = palloc(sizeof(int32));
+
+	/* initialize with the default acceleration */
+	*acceleration = 1;
+
+	if (list_length(options) > 0)
+	{
+		ListCell	*lc;
+
+		foreach(lc, options)
+		{
+			DefElem    *def = (DefElem *) lfirst(lc);
+
+			if (strcmp(def->defname, "acceleration") == 0)
+				*acceleration = pg_atoi(defGetString(def), sizeof(int32), 0);
+		}
+	}
+
+	return acceleration;
+}
+
 /*
  * lz4_cmcompress - compression routine for lz4 compression method
  *
@@ -29,7 +87,7 @@
  * compressed varlena, or NULL if compression fails.
  */
 static struct varlena *
-lz4_cmcompress(const struct varlena *value, int32 header_size)
+lz4_cmcompress(const struct varlena *value, int32 header_size, void *options)
 {
 #ifndef HAVE_LIBLZ4
 	ereport(ERROR,
@@ -39,6 +97,7 @@ lz4_cmcompress(const struct varlena *value, int32 header_size)
 	int32		valsize;
 	int32		len;
 	int32		max_size;
+	int32      *acceleration = (int32 *) options;
 	struct varlena *tmp = NULL;
 
 	valsize = VARSIZE_ANY_EXHDR(value);
@@ -46,9 +105,9 @@ lz4_cmcompress(const struct varlena *value, int32 header_size)
 	max_size = LZ4_compressBound(VARSIZE_ANY_EXHDR(value));
 	tmp = (struct varlena *) palloc(max_size + header_size);
 
-	len = LZ4_compress_default(VARDATA_ANY(value),
-							   (char *) tmp + header_size,
-							   valsize, max_size);
+	len = LZ4_compress_fast(VARDATA_ANY(value),
+							(char *) tmp + header_size,
+							valsize, max_size, *acceleration);
 	if (len <= 0)
 	{
 		pfree(tmp);
@@ -129,6 +188,8 @@ lz4_cmdecompress_slice(const struct varlena *value, int32 header_size,
 
 const CompressionAmRoutine lz4_compress_methods = {
 	.type = T_CompressionAmRoutine,
+	.datum_check = lz4_cmcheck,
+	.datum_initstate = lz4_cminitstate,
 	.datum_compress = lz4_cmcompress,
 	.datum_decompress = lz4_cmdecompress,
 	.datum_decompress_slice = lz4_cmdecompress_slice
diff --git a/src/backend/access/compression/compress_pglz.c b/src/backend/access/compression/compress_pglz.c
index d693c880fd..7cc5a22e37 100644
--- a/src/backend/access/compression/compress_pglz.c
+++ b/src/backend/access/compression/compress_pglz.c
@@ -15,11 +15,92 @@
 
 #include "access/compressamapi.h"
 #include "access/toast_internals.h"
+#include "commands/defrem.h"
 #include "common/pg_lzcompress.h"
 
 #include "fmgr.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(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(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;
+}
+
 /*
  * pglz_cmcompress - compression routine for pglz compression method
  *
@@ -27,20 +108,22 @@
  * compressed varlena, or NULL if compression fails.
  */
 static struct varlena *
-pglz_cmcompress(const struct varlena *value, int32 header_size)
+pglz_cmcompress(const struct varlena *value, int32 header_size, void *options)
 {
 	int32		valsize,
 				len;
-	struct varlena *tmp = NULL;
+	struct varlena	*tmp = NULL;
+	PGLZ_Strategy	*strategy;
 
 	valsize = VARSIZE_ANY_EXHDR(DatumGetPointer(value));
+	strategy = (PGLZ_Strategy *) options;
 
 	/*
 	 * No point in wasting a palloc cycle if value size is out of the allowed
 	 * range for compression
 	 */
-	if (valsize < PGLZ_strategy_default->min_input_size ||
-		valsize > PGLZ_strategy_default->max_input_size)
+	if (valsize < strategy->min_input_size ||
+		valsize > strategy->max_input_size)
 		return NULL;
 
 	tmp = (struct varlena *) palloc(PGLZ_MAX_OUTPUT(valsize) +
@@ -49,7 +132,7 @@ pglz_cmcompress(const struct varlena *value, int32 header_size)
 	len = pglz_compress(VARDATA_ANY(DatumGetPointer(value)),
 						valsize,
 						(char *) tmp + header_size,
-						NULL);
+						strategy);
 
 	if (len >= 0)
 	{
@@ -118,10 +201,11 @@ pglz_cmdecompress_slice(const struct varlena *value, int32 header_size,
 
 const CompressionAmRoutine pglz_compress_methods = {
 	.type = T_CompressionAmRoutine,
+	.datum_check = pglz_cmcheck,
+	.datum_initstate = pglz_cminitstate,
 	.datum_compress = pglz_cmcompress,
 	.datum_decompress = pglz_cmdecompress,
-	.datum_decompress_slice = pglz_cmdecompress_slice
-};
+	.datum_decompress_slice = pglz_cmdecompress_slice};
 
 /* pglz compression handler function */
 Datum
diff --git a/src/backend/access/table/toast_helper.c b/src/backend/access/table/toast_helper.c
index b33c925a4b..7e399daaa1 100644
--- a/src/backend/access/table/toast_helper.c
+++ b/src/backend/access/table/toast_helper.c
@@ -15,11 +15,13 @@
 #include "postgres.h"
 
 #include "access/detoast.h"
+#include "access/reloptions.h"
 #include "access/table.h"
 #include "access/toast_helper.h"
 #include "access/toast_internals.h"
 #include "catalog/pg_type_d.h"
-
+#include "commands/defrem.h"
+#include "utils/syscache.h"
 
 /*
  * Prepare to TOAST a tuple.
@@ -55,6 +57,7 @@ toast_tuple_init(ToastTupleContext *ttc)
 		ttc->ttc_attr[i].tai_colflags = 0;
 		ttc->ttc_attr[i].tai_oldexternal = NULL;
 		ttc->ttc_attr[i].tai_compression = att->attcompression;
+		ttc->ttc_attr[i].tai_cmoptions = GetAttributeCompressionOptions(att);
 
 		if (ttc->ttc_oldvalues != NULL)
 		{
@@ -230,7 +233,8 @@ toast_tuple_try_compression(ToastTupleContext *ttc, int attribute)
 	Datum		new_value;
 	ToastAttrInfo *attr = &ttc->ttc_attr[attribute];
 
-	new_value = toast_compress_datum(*value, attr->tai_compression);
+	new_value = toast_compress_datum(*value, attr->tai_compression,
+									 attr->tai_cmoptions);
 
 	if (DatumGetPointer(new_value) != NULL)
 	{
diff --git a/src/backend/bootstrap/bootparse.y b/src/backend/bootstrap/bootparse.y
index 6bb0c6ed1e..bbc849b541 100644
--- a/src/backend/bootstrap/bootparse.y
+++ b/src/backend/bootstrap/bootparse.y
@@ -239,6 +239,7 @@ Boot_CreateStmt:
 													  true,
 													  false,
 													  InvalidOid,
+													  NULL,
 													  NULL);
 						elog(DEBUG4, "relation created with OID %u", id);
 					}
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index c2189ef0ef..708a484880 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -732,6 +732,7 @@ InsertPgAttributeTuples(Relation pg_attribute_rel,
 						TupleDesc tupdesc,
 						Oid new_rel_oid,
 						Datum *attoptions,
+						Datum *attcmoptions,
 						CatalogIndexState indstate)
 {
 	TupleTableSlot **slot;
@@ -787,6 +788,11 @@ InsertPgAttributeTuples(Relation pg_attribute_rel,
 		else
 			slot[slotCount]->tts_isnull[Anum_pg_attribute_attoptions - 1] = true;
 
+		if (attcmoptions && attcmoptions[natts] != (Datum) 0)
+			slot[slotCount]->tts_values[Anum_pg_attribute_attcmoptions - 1] = attcmoptions[natts];
+		else
+			slot[slotCount]->tts_isnull[Anum_pg_attribute_attcmoptions - 1] = true;
+
 		/* start out with empty permissions and empty options */
 		slot[slotCount]->tts_isnull[Anum_pg_attribute_attacl - 1] = true;
 		slot[slotCount]->tts_isnull[Anum_pg_attribute_attfdwoptions - 1] = true;
@@ -834,6 +840,7 @@ InsertPgAttributeTuples(Relation pg_attribute_rel,
 static void
 AddNewAttributeTuples(Oid new_rel_oid,
 					  TupleDesc tupdesc,
+					  Datum *acoption,
 					  char relkind)
 {
 	Relation	rel;
@@ -852,7 +859,8 @@ AddNewAttributeTuples(Oid new_rel_oid,
 	/* set stats detail level to a sane default */
 	for (int i = 0; i < natts; i++)
 		tupdesc->attrs[i].attstattarget = -1;
-	InsertPgAttributeTuples(rel, tupdesc, new_rel_oid, NULL, indstate);
+	InsertPgAttributeTuples(rel, tupdesc, new_rel_oid,
+							NULL, acoption, indstate);
 
 	/* add dependencies on their datatypes and collations */
 	for (int i = 0; i < natts; i++)
@@ -884,7 +892,7 @@ AddNewAttributeTuples(Oid new_rel_oid,
 
 		td = CreateTupleDesc(lengthof(SysAtt), (FormData_pg_attribute **) &SysAtt);
 
-		InsertPgAttributeTuples(rel, td, new_rel_oid, NULL, indstate);
+		InsertPgAttributeTuples(rel, td, new_rel_oid, NULL, NULL, indstate);
 		FreeTupleDesc(td);
 	}
 
@@ -1150,6 +1158,7 @@ heap_create_with_catalog(const char *relname,
 						 bool allow_system_table_mods,
 						 bool is_internal,
 						 Oid relrewrite,
+						 Datum *acoptions,
 						 ObjectAddress *typaddress)
 {
 	Relation	pg_class_desc;
@@ -1407,7 +1416,7 @@ heap_create_with_catalog(const char *relname,
 	/*
 	 * now add tuples to pg_attribute for the attributes in our new relation.
 	 */
-	AddNewAttributeTuples(relid, new_rel_desc->rd_att, relkind);
+	AddNewAttributeTuples(relid, new_rel_desc->rd_att, acoptions, relkind);
 
 	/*
 	 * Make a dependency link to force the relation to be deleted if its
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index 7e6310b6f7..b878b4af7f 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -107,10 +107,12 @@ static TupleDesc ConstructTupleDescriptor(Relation heapRelation,
 										  List *indexColNames,
 										  Oid accessMethodObjectId,
 										  Oid *collationObjectId,
-										  Oid *classObjectId);
+										  Oid *classObjectId,
+										  Datum *acoptions);
 static void InitializeAttributeOids(Relation indexRelation,
 									int numatts, Oid indexoid);
-static void AppendAttributeTuples(Relation indexRelation, Datum *attopts);
+static void AppendAttributeTuples(Relation indexRelation, Datum *attopts,
+								  Datum *attcmopts);
 static void UpdateIndexRelation(Oid indexoid, Oid heapoid,
 								Oid parentIndexId,
 								IndexInfo *indexInfo,
@@ -272,7 +274,8 @@ ConstructTupleDescriptor(Relation heapRelation,
 						 List *indexColNames,
 						 Oid accessMethodObjectId,
 						 Oid *collationObjectId,
-						 Oid *classObjectId)
+						 Oid *classObjectId,
+						 Datum *acoptions)
 {
 	int			numatts = indexInfo->ii_NumIndexAttrs;
 	int			numkeyatts = indexInfo->ii_NumIndexKeyAttrs;
@@ -333,6 +336,9 @@ ConstructTupleDescriptor(Relation heapRelation,
 		{
 			/* Simple index column */
 			const FormData_pg_attribute *from;
+			HeapTuple	attr_tuple;
+			Datum		attcmoptions;
+			bool		isNull;
 
 			Assert(atnum > 0);	/* should've been caught above */
 
@@ -349,6 +355,23 @@ ConstructTupleDescriptor(Relation heapRelation,
 			to->attstorage = from->attstorage;
 			to->attalign = from->attalign;
 			to->attcompression = from->attcompression;
+
+			attr_tuple = SearchSysCache2(ATTNUM,
+										 ObjectIdGetDatum(from->attrelid),
+										 Int16GetDatum(from->attnum));
+			if (!HeapTupleIsValid(attr_tuple))
+				elog(ERROR, "cache lookup failed for attribute %d of relation %u",
+					 from->attnum, from->attrelid);
+
+			attcmoptions = SysCacheGetAttr(ATTNUM, attr_tuple,
+										   Anum_pg_attribute_attcmoptions,
+										   &isNull);
+			if (isNull)
+				acoptions[i] = PointerGetDatum(NULL);
+			else
+				acoptions[i] =
+					PointerGetDatum(DatumGetArrayTypePCopy(attcmoptions));
+			ReleaseSysCache(attr_tuple);
 		}
 		else
 		{
@@ -489,7 +512,7 @@ InitializeAttributeOids(Relation indexRelation,
  * ----------------------------------------------------------------
  */
 static void
-AppendAttributeTuples(Relation indexRelation, Datum *attopts)
+AppendAttributeTuples(Relation indexRelation, Datum *attopts, Datum *attcmopts)
 {
 	Relation	pg_attribute;
 	CatalogIndexState indstate;
@@ -507,7 +530,8 @@ AppendAttributeTuples(Relation indexRelation, Datum *attopts)
 	 */
 	indexTupDesc = RelationGetDescr(indexRelation);
 
-	InsertPgAttributeTuples(pg_attribute, indexTupDesc, InvalidOid, attopts, indstate);
+	InsertPgAttributeTuples(pg_attribute, indexTupDesc, InvalidOid,
+							attopts, attcmopts, indstate);
 
 	CatalogCloseIndexes(indstate);
 
@@ -720,6 +744,7 @@ index_create(Relation heapRelation,
 	bool		concurrent = (flags & INDEX_CREATE_CONCURRENT) != 0;
 	bool		partitioned = (flags & INDEX_CREATE_PARTITIONED) != 0;
 	char		relkind;
+	Datum	   *acoptions;
 	TransactionId relfrozenxid;
 	MultiXactId relminmxid;
 
@@ -875,6 +900,8 @@ index_create(Relation heapRelation,
 						indexRelationName, RelationGetRelationName(heapRelation))));
 	}
 
+	acoptions = (Datum *) palloc0(sizeof(Datum) * indexInfo->ii_NumIndexAttrs);
+
 	/*
 	 * construct tuple descriptor for index tuples
 	 */
@@ -883,7 +910,8 @@ index_create(Relation heapRelation,
 											indexColNames,
 											accessMethodObjectId,
 											collationObjectId,
-											classObjectId);
+											classObjectId,
+											acoptions);
 
 	/*
 	 * Allocate an OID for the index, unless we were told what to use.
@@ -974,7 +1002,8 @@ index_create(Relation heapRelation,
 	/*
 	 * append ATTRIBUTE tuples for the index
 	 */
-	AppendAttributeTuples(indexRelation, indexInfo->ii_OpclassOptions);
+	AppendAttributeTuples(indexRelation, indexInfo->ii_OpclassOptions,
+						  acoptions);
 
 	/* ----------------
 	 *	  update pg_index
diff --git a/src/backend/catalog/toasting.c b/src/backend/catalog/toasting.c
index e5452614f7..6cf32fef6a 100644
--- a/src/backend/catalog/toasting.c
+++ b/src/backend/catalog/toasting.c
@@ -260,6 +260,7 @@ create_toast_table(Relation rel, Oid toastOid, Oid toastIndexOid,
 										   true,
 										   true,
 										   InvalidOid,
+										   NULL,
 										   NULL);
 	Assert(toast_relid != InvalidOid);
 
diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c
index fd5a6eec86..29a8145b1d 100644
--- a/src/backend/commands/cluster.c
+++ b/src/backend/commands/cluster.c
@@ -697,6 +697,7 @@ make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, char relpersistence,
 										  true,
 										  true,
 										  OIDOldHeap,
+										  NULL,
 										  NULL);
 	Assert(OIDNewHeap != InvalidOid);
 
diff --git a/src/backend/commands/compressioncmds.c b/src/backend/commands/compressioncmds.c
index 91eb59ca17..934345be1f 100644
--- a/src/backend/commands/compressioncmds.c
+++ b/src/backend/commands/compressioncmds.c
@@ -27,6 +27,7 @@
 #include "utils/builtins.h"
 #include "utils/fmgroids.h"
 #include "utils/lsyscache.h"
+#include "utils/syscache.h"
 
 /*
  * get list of all supported compression methods for the given attribute.
@@ -127,7 +128,7 @@ IsCompressionSupported(Form_pg_attribute att, Oid cmoid)
  */
 Oid
 GetAttributeCompression(Form_pg_attribute att, ColumnCompression *compression,
-						bool *need_rewrite)
+						Datum *acoptions, bool *need_rewrite)
 {
 	Oid			cmoid;
 	char		typstorage = get_typstorage(att->atttypid);
@@ -153,6 +154,20 @@ GetAttributeCompression(Form_pg_attribute att, ColumnCompression *compression,
 
 	cmoid = get_compression_am_oid(compression->cmname, false);
 
+	/* if compression options are given then check them */
+	if (compression->options)
+	{
+		CompressionAmRoutine *routine = GetCompressionAmRoutineByAmId(cmoid);
+
+		/* we need routine only to call cmcheck function */
+		if (routine->datum_check != NULL)
+			routine->datum_check(compression->options);
+
+		*acoptions = optionListToArray(compression->options);
+	}
+	else
+		*acoptions = PointerGetDatum(NULL);
+
 	/*
 	 * Determine if the column needs rewrite or not. Rewrite conditions: SET
 	 * COMPRESSION without PRESERVE - SET COMPRESSION with PRESERVE but not
@@ -229,3 +244,34 @@ MakeColumnCompression(Oid attcompression)
 
 	return node;
 }
+
+/*
+ * Fetch atttributes compression options
+ */
+List *
+GetAttributeCompressionOptions(Form_pg_attribute att)
+{
+	HeapTuple	attr_tuple;
+	Datum		attcmoptions;
+	List	   *acoptions;
+	bool		isNull;
+
+	attr_tuple = SearchSysCache2(ATTNUM,
+								 ObjectIdGetDatum(att->attrelid),
+								 Int16GetDatum(att->attnum));
+	if (!HeapTupleIsValid(attr_tuple))
+		elog(ERROR, "cache lookup failed for attribute %d of relation %u",
+			 att->attnum, att->attrelid);
+
+	attcmoptions = SysCacheGetAttr(ATTNUM, attr_tuple,
+								   Anum_pg_attribute_attcmoptions,
+								   &isNull);
+	if (isNull)
+		acoptions = NULL;
+	else
+		acoptions = untransformRelOptions(attcmoptions);
+
+	ReleaseSysCache(attr_tuple);
+
+	return acoptions;
+}
diff --git a/src/backend/commands/foreigncmds.c b/src/backend/commands/foreigncmds.c
index de31ddd1f3..95bdcb1874 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
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 91a341fc3e..2df9e52c49 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -611,6 +611,7 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 	LOCKMODE	parentLockmode;
 	const char *accessMethod = NULL;
 	Oid			accessMethodId = InvalidOid;
+	Datum	   *acoptions;
 
 	/*
 	 * Truncate relname to appropriate length (probably a waste of time, as
@@ -815,6 +816,8 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 	cookedDefaults = NIL;
 	attnum = 0;
 
+	acoptions = (Datum *) palloc0(sizeof(Datum) * descriptor->natts);
+
 	foreach(listptr, stmt->tableElts)
 	{
 		ColumnDef  *colDef = lfirst(listptr);
@@ -868,8 +871,9 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 		if (relkind == RELKIND_RELATION ||
 			relkind == RELKIND_PARTITIONED_TABLE ||
 			relkind == RELKIND_MATVIEW)
-			attr->attcompression =
-				GetAttributeCompression(attr, colDef->compression, NULL);
+			attr->attcompression = GetAttributeCompression(attr,
+												colDef->compression,
+												&acoptions[attnum - 1], NULL);
 		else
 			attr->attcompression = InvalidOid;
 	}
@@ -923,8 +927,11 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 										  allowSystemTableMods,
 										  false,
 										  InvalidOid,
+										  acoptions,
 										  typaddress);
 
+	pfree(acoptions);
+
 	/*
 	 * We must bump the command counter to make the newly-created relation
 	 * tuple visible for opening.
@@ -6153,6 +6160,7 @@ ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel,
 	AclResult	aclresult;
 	ObjectAddress address;
 	TupleDesc	tupdesc;
+	Datum		acoptions = PointerGetDatum(NULL);
 	FormData_pg_attribute *aattr[] = {&attribute};
 
 	/* At top level, permission check was done in ATPrepCmd, else do it */
@@ -6316,6 +6324,7 @@ ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel,
 		rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
 		attribute.attcompression = GetAttributeCompression(&attribute,
 														   colDef->compression,
+														   &acoptions,
 														   NULL);
 	else
 		attribute.attcompression = InvalidOid;
@@ -6326,7 +6335,7 @@ ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel,
 
 	tupdesc = CreateTupleDesc(lengthof(aattr), (FormData_pg_attribute **) &aattr);
 
-	InsertPgAttributeTuples(attrdesc, tupdesc, myrelid, NULL, NULL);
+	InsertPgAttributeTuples(attrdesc, tupdesc, myrelid, NULL, &acoptions, NULL);
 
 	table_close(attrdesc, RowExclusiveLock);
 
@@ -7723,13 +7732,20 @@ ATExecSetOptions(Relation rel, const char *colName, Node *options,
  * index attribute otherwise it will update the attstorage.
  */
 static void
-ApplyChangesToIndexes(Relation rel, Relation attrelation, AttrNumber attnum,
-					  Oid newcompression, char newstorage, LOCKMODE lockmode)
+ApplyChangesToIndexes(Relation rel, Relation attrel, AttrNumber attnum,
+					  Oid newcompression, char newstorage, Datum acoptions,
+					  LOCKMODE lockmode)
 {
 	HeapTuple	tuple;
 	ListCell   *lc;
 	Form_pg_attribute attrtuple;
 
+	/*
+	 * Compression option must be only valid if we are updating the compression
+	 * method.
+	 */
+	Assert(DatumGetPointer(acoptions) == NULL || OidIsValid(newcompression));
+
 	foreach(lc, RelationGetIndexList(rel))
 	{
 		Oid			indexoid = lfirst_oid(lc);
@@ -7768,7 +7784,29 @@ ApplyChangesToIndexes(Relation rel, Relation attrelation, AttrNumber attnum,
 			else
 				attrtuple->attstorage = newstorage;
 
-			CatalogTupleUpdate(attrelation, &tuple->t_self, tuple);
+			if (DatumGetPointer(acoptions) != NULL)
+			{
+				Datum	values[Natts_pg_attribute];
+				bool	nulls[Natts_pg_attribute];
+				bool	replace[Natts_pg_attribute];
+				HeapTuple	newtuple;
+
+				/* Initialize buffers for new tuple values */
+				memset(values, 0, sizeof(values));
+				memset(nulls, false, sizeof(nulls));
+				memset(replace, false, sizeof(replace));
+
+				values[Anum_pg_attribute_attcmoptions - 1] = acoptions;
+				replace[Anum_pg_attribute_attcmoptions - 1] = true;
+
+				newtuple = heap_modify_tuple(tuple, RelationGetDescr(attrel),
+											 values, nulls, replace);
+				CatalogTupleUpdate(attrel, &newtuple->t_self, newtuple);
+
+				heap_freetuple(newtuple);
+			}
+			else
+				CatalogTupleUpdate(attrel, &tuple->t_self, tuple);
 
 			InvokeObjectPostAlterHook(RelationRelationId,
 									  RelationGetRelid(rel),
@@ -7859,7 +7897,7 @@ ATExecSetStorage(Relation rel, const char *colName, Node *newValue, LOCKMODE loc
 	 * matching behavior of index.c ConstructTupleDescriptor()).
 	 */
 	ApplyChangesToIndexes(rel, attrelation, attnum, InvalidOid, newstorage,
-						  lockmode);
+						  PointerGetDatum(NULL), lockmode);
 
 	table_close(attrelation, RowExclusiveLock);
 
@@ -15105,9 +15143,11 @@ ATExecSetCompression(AlteredTableInfo *tab,
 	char		typstorage;
 	Oid			cmoid;
 	bool		need_rewrite;
+	HeapTuple	newtuple;
 	Datum		values[Natts_pg_attribute];
 	bool		nulls[Natts_pg_attribute];
 	bool		replace[Natts_pg_attribute];
+	Datum		acoptions;
 	ObjectAddress address;
 
 	attrel = table_open(AttributeRelationId, RowExclusiveLock);
@@ -15142,7 +15182,8 @@ ATExecSetCompression(AlteredTableInfo *tab,
 	memset(replace, false, sizeof(replace));
 
 	/* Get the attribute compression method. */
-	cmoid = GetAttributeCompression(atttableform, compression, &need_rewrite);
+	cmoid = GetAttributeCompression(atttableform, compression, &acoptions,
+									&need_rewrite);
 
 	if (atttableform->attcompression != cmoid)
 		add_column_compression_dependency(atttableform->attrelid,
@@ -15152,8 +15193,17 @@ ATExecSetCompression(AlteredTableInfo *tab,
 
 	atttableform->attcompression = cmoid;
 
-	atttableform->attcompression = cmoid;
-	CatalogTupleUpdate(attrel, &tuple->t_self, tuple);
+	/* update an existing entry */
+	if (acoptions)
+	{
+		values[Anum_pg_attribute_attcmoptions - 1] = acoptions;
+		replace[Anum_pg_attribute_attcmoptions - 1] = true;
+		newtuple = heap_modify_tuple(tuple, RelationGetDescr(attrel),
+									 values, nulls, replace);
+		CatalogTupleUpdate(attrel, &newtuple->t_self, newtuple);
+	}
+	else
+		CatalogTupleUpdate(attrel, &tuple->t_self, tuple);
 
 	InvokeObjectPostAlterHook(RelationRelationId,
 							  RelationGetRelid(rel),
@@ -15161,8 +15211,67 @@ ATExecSetCompression(AlteredTableInfo *tab,
 
 	ReleaseSysCache(tuple);
 
-	/* apply changes to the index column as well */
-	ApplyChangesToIndexes(rel, attrel, attnum, cmoid, '\0', lockmode);
+	/* Apply the change to indexes as well */
+	ApplyChangesToIndexes(rel, attrel, attnum, cmoid, '\0', acoptions,
+						  lockmode);
+#if 0						  
+	foreach (lc, RelationGetIndexList(rel))
+	{
+		Oid indexoid = lfirst_oid(lc);
+		Relation indrel;
+		AttrNumber indattnum = 0;
+
+		indrel = index_open(indexoid, lockmode);
+
+		for (int i = 0; i < indrel->rd_index->indnatts; i++)
+		{
+			if (indrel->rd_index->indkey.values[i] == attnum)
+			{
+				indattnum = i + 1;
+				break;
+			}
+		}
+
+		if (indattnum == 0)
+		{
+			index_close(indrel, lockmode);
+			continue;
+		}
+
+		tuple = SearchSysCacheCopyAttNum(RelationGetRelid(indrel), indattnum);
+
+		if (HeapTupleIsValid(tuple))
+		{
+			atttableform = (Form_pg_attribute) GETSTRUCT(tuple);
+			atttableform->attcompression = cmoid;
+
+			/* update existing entry */
+			if (acoptions)
+			{
+				/* Initialize buffers for new tuple values */
+				memset(values, 0, sizeof(values));
+				memset(nulls, false, sizeof(nulls));
+				memset(replace, false, sizeof(replace));
+
+				values[Anum_pg_attribute_attcmoptions - 1] = PointerGetDatum(acoptions);
+				replace[Anum_pg_attribute_attcmoptions - 1] = true;
+				newtuple = heap_modify_tuple(tuple, RelationGetDescr(attrel),
+											 values, nulls, replace);
+				CatalogTupleUpdate(attrel, &newtuple->t_self, newtuple);
+			}
+			else
+				CatalogTupleUpdate(attrel, &tuple->t_self, tuple);
+
+			InvokeObjectPostAlterHook(RelationRelationId,
+									  RelationGetRelid(rel),
+									  atttableform->attnum);
+
+			heap_freetuple(tuple);
+		}
+
+		index_close(indrel, lockmode);
+	}
+#endif	
 	table_close(attrel, RowExclusiveLock);
 
 	/* make changes visible */
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 2ac6d61320..2436abd467 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -2958,6 +2958,7 @@ _copyColumnCompression(const ColumnCompression *from)
 
 	COPY_STRING_FIELD(cmname);
 	COPY_SCALAR_FIELD(preserve_all);
+	COPY_NODE_FIELD(options);
 	COPY_NODE_FIELD(preserve);
 
 	return newnode;
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index 51dee9e12f..ae6cfb0266 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -2610,6 +2610,7 @@ _equalColumnCompression(const ColumnCompression *a, const ColumnCompression *b)
 {
 	COMPARE_STRING_FIELD(cmname);
 	COMPARE_SCALAR_FIELD(preserve_all);
+	COMPARE_NODE_FIELD(options);
 	COMPARE_NODE_FIELD(preserve);
 
 	return true;
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index f458acae6d..323bccd345 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -2877,6 +2877,7 @@ _outColumnCompression(StringInfo str, const ColumnCompression *node)
 
 	WRITE_STRING_FIELD(cmname);
 	WRITE_BOOL_FIELD(preserve_all);
+	WRITE_NODE_FIELD(options);
 	WRITE_NODE_FIELD(preserve);
 }
 
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 54ff6fb937..65ea7a4d39 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -413,6 +413,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 				relation_expr_list dostmt_opt_list
 				transform_element_list transform_type_list
 				TriggerTransitions TriggerReferencing
+				optCompressionParameters
 				vacuum_relation_list opt_vacuum_relation_list
 				drop_option_list
 
@@ -3452,11 +3453,17 @@ compressionClause:
 			COMPRESSION name { $$ = pstrdup($2); }
 		;
 
+optCompressionParameters:
+			WITH '(' generic_option_list ')' { $$ = $3; }
+			| /*EMPTY*/	{ $$ = NULL; }
+		;
+
 optColumnCompression:
-			compressionClause
+			compressionClause optCompressionParameters
 				{
 					ColumnCompression *n = makeNode(ColumnCompression);
 					n->cmname = $1;
+					n->options = (List *) $2;
 					n->preserve = NIL;
 					$$ = (Node *) n;
 				}
@@ -3464,14 +3471,15 @@ optColumnCompression:
 		;
 
 alterColumnCompression:
-			compressionClause optCompressionPreserve
+			compressionClause optCompressionParameters optCompressionPreserve
 				{
 					ColumnCompression *n = makeNode(ColumnCompression);
 					n->cmname = $1;
-					n->preserve = (List *) $2;
+					n->options = (List *) $2;
+					n->preserve = (List *) $3;
 					$$ = (Node *) n;
 				}
-			|	compressionClause PRESERVE ALL
+			|	compressionClause optCompressionParameters PRESERVE ALL
 				{
 					ColumnCompression *n = makeNode(ColumnCompression);
 					n->cmname = $1;
diff --git a/src/include/access/compressamapi.h b/src/include/access/compressamapi.h
index 913a633d83..8123cc8cc7 100644
--- a/src/include/access/compressamapi.h
+++ b/src/include/access/compressamapi.h
@@ -17,6 +17,7 @@
 
 #include "catalog/pg_am_d.h"
 #include "nodes/nodes.h"
+#include "nodes/pg_list.h"
 
 /*
  * Built-in compression method-id.  The toast compression header will store
@@ -37,8 +38,11 @@ typedef enum CompressionId
 #define IsStorageCompressible(storage) ((storage) != TYPSTORAGE_PLAIN && \
 										(storage) != TYPSTORAGE_EXTERNAL)
 /* compression handler routines */
+typedef void (*cmcheck_function) (List *options);
+typedef void *(*cminitstate_function) (List *options);
 typedef struct varlena *(*cmcompress_function) (const struct varlena *value,
-												int32 toast_header_size);
+												int32 toast_header_size,
+												void *options);
 typedef struct varlena *(*cmdecompress_function) (const struct varlena *value,
 												  int32 toast_header_size);
 typedef struct varlena *(*cmdecompress_slice_function)
@@ -49,14 +53,25 @@ typedef struct varlena *(*cmdecompress_slice_function)
 /*
  * API struct for a compression AM.
  *
+ * 'cmcheck' - called when attribute is linking with compression method.
+ *  This function should check compability of compression method with
+ *  the attribute and its options.
+ *
+ * 'cminitstate' - called when CompressionAmOptions instance is created.
+ *  Should return pointer to a memory in a caller memory context, or NULL.
+ *  Could be used to pass some internal state between compression function
+ *  calls, like internal structure for parsed compression options.
+ *
  * 'datum_compress' - varlena compression function.
  * 'datum_decompress' - varlena decompression function.
  * 'datum_decompress_slice' - varlena slice decompression functions.
  */
 typedef struct CompressionAmRoutine
 {
-	NodeTag		type;
+	NodeTag type;
 
+	cmcheck_function datum_check;		  /* can be NULL */
+	cminitstate_function datum_initstate; /* can be NULL */
 	cmcompress_function datum_compress;
 	cmdecompress_function datum_decompress;
 	cmdecompress_slice_function datum_decompress_slice;
diff --git a/src/include/access/toast_helper.h b/src/include/access/toast_helper.h
index 4e932bc067..1dc4aa9623 100644
--- a/src/include/access/toast_helper.h
+++ b/src/include/access/toast_helper.h
@@ -15,6 +15,7 @@
 #define TOAST_HELPER_H
 
 #include "utils/rel.h"
+#include "nodes/pg_list.h"
 
 /*
  * Information about one column of a tuple being toasted.
@@ -33,6 +34,7 @@ typedef struct
 	int32		tai_size;
 	uint8		tai_colflags;
 	Oid			tai_compression;
+	List	   *tai_cmoptions;
 } ToastAttrInfo;
 
 /*
diff --git a/src/include/access/toast_internals.h b/src/include/access/toast_internals.h
index 81944912d2..dc35064086 100644
--- a/src/include/access/toast_internals.h
+++ b/src/include/access/toast_internals.h
@@ -64,7 +64,7 @@ typedef struct toast_compress_header_custom
 #define TOAST_COMPRESS_SET_CMOID(ptr, oid) \
 	(((toast_compress_header_custom *)(ptr))->cmoid = (oid))
 
-extern Datum toast_compress_datum(Datum value, Oid cmoid);
+extern Datum toast_compress_datum(Datum value, Oid cmoid, List *cmoptions);
 extern Oid	toast_get_valid_index(Oid toastoid, LOCKMODE lock);
 
 extern void toast_delete_datum(Relation rel, Datum value, bool is_speculative);
diff --git a/src/include/catalog/heap.h b/src/include/catalog/heap.h
index f94defff3c..bd192f457e 100644
--- a/src/include/catalog/heap.h
+++ b/src/include/catalog/heap.h
@@ -81,6 +81,7 @@ extern Oid	heap_create_with_catalog(const char *relname,
 									 bool allow_system_table_mods,
 									 bool is_internal,
 									 Oid relrewrite,
+									 Datum *acoptions,
 									 ObjectAddress *typaddress);
 
 extern void heap_drop_with_catalog(Oid relid);
@@ -97,6 +98,7 @@ extern void InsertPgAttributeTuples(Relation pg_attribute_rel,
 									TupleDesc tupdesc,
 									Oid new_rel_oid,
 									Datum *attoptions,
+									Datum *attcmoptions,
 									CatalogIndexState indstate);
 
 extern void InsertPgClassTuple(Relation pg_class_desc,
diff --git a/src/include/catalog/pg_attribute.h b/src/include/catalog/pg_attribute.h
index 3e6ed706ac..55ea9af6bd 100644
--- a/src/include/catalog/pg_attribute.h
+++ b/src/include/catalog/pg_attribute.h
@@ -176,6 +176,9 @@ CATALOG(pg_attribute,1249,AttributeRelationId) BKI_BOOTSTRAP BKI_ROWTYPE_OID(75,
 	/* Column-level FDW options */
 	text		attfdwoptions[1] BKI_DEFAULT(_null_);
 
+	/* current compression options */
+	text		attcmoptions[1] BKI_DEFAULT(_null_);
+
 	/*
 	 * Missing value for added columns. This is a one element array which lets
 	 * us store a value of the attribute type here.
diff --git a/src/include/commands/defrem.h b/src/include/commands/defrem.h
index d1880dfc5e..b593103367 100644
--- a/src/include/commands/defrem.h
+++ b/src/include/commands/defrem.h
@@ -138,6 +138,7 @@ extern Datum transformGenericOptions(Oid catalogId,
 									 Datum oldOptions,
 									 List *options,
 									 Oid fdwvalidator);
+extern Datum optionListToArray(List *options);
 
 /* commands/amcmds.c */
 extern ObjectAddress CreateAccessMethod(CreateAmStmt *stmt);
@@ -150,9 +151,10 @@ extern char *get_am_name(Oid amOid);
 /* commands/compressioncmds.c */
 extern Oid GetAttributeCompression(Form_pg_attribute att,
 								   ColumnCompression *compression,
-								   bool *need_rewrite);
+								   Datum *acoptions, bool *need_rewrite);
 extern ColumnCompression *MakeColumnCompression(Oid atttcompression);
 extern bool IsCompressionSupported(Form_pg_attribute att, Oid cmoid);
+extern List *GetAttributeCompressionOptions(Form_pg_attribute att);
 
 /* support routines in commands/define.c */
 
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index c8135debbb..09e5e00d99 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -634,6 +634,7 @@ typedef struct ColumnCompression
 	NodeTag		type;
 	char	   *cmname;
 	bool		preserve_all;
+	List	   *options;
 	List	   *preserve;
 } ColumnCompression;
 
diff --git a/src/test/regress/expected/compression.out b/src/test/regress/expected/compression.out
index 2c336efaed..0f977c5e92 100644
--- a/src/test/regress/expected/compression.out
+++ b/src/test/regress/expected/compression.out
@@ -263,6 +263,23 @@ SELECT pg_column_compression(f1) FROM cmdata;
 Indexes:
     "idx" btree (f1)
 
+-- compression options
+DROP TABLE cmmove1;
+CREATE TABLE cmmove1(f1 TEXT COMPRESSION pglz WITH (min_input_size '100'));
+CREATE INDEX idx1 ON cmdata(f1);
+INSERT INTO cmmove1 VALUES(repeat('1234567890',1000));
+ALTER TABLE cmmove1 ALTER COLUMN f1 SET COMPRESSION lz4 WITH (acceleration '50');
+INSERT INTO cmmove1 VALUES(repeat('1234567890',1004));
+ALTER TABLE cmmove1 ALTER COLUMN f1 SET COMPRESSION pglz WITH (min_input_size '200') PRESERVE (lz4);
+INSERT INTO cmmove1 VALUES(repeat('1234567890',1008));
+SELECT pg_column_compression(f1) FROM cmmove1;
+ pg_column_compression 
+-----------------------
+ lz4
+ lz4
+ pglz
+(3 rows)
+
 -- check data is ok
 SELECT length(f1) FROM cmdata;
  length 
@@ -282,7 +299,9 @@ SELECT length(f1) FROM cmmove1;
  length 
 --------
   10000
-(1 row)
+  10040
+  10080
+(3 rows)
 
 SELECT length(f1) FROM cmmove2;
  length 
diff --git a/src/test/regress/expected/misc_sanity.out b/src/test/regress/expected/misc_sanity.out
index d40afeef78..27aab82462 100644
--- a/src/test/regress/expected/misc_sanity.out
+++ b/src/test/regress/expected/misc_sanity.out
@@ -97,6 +97,7 @@ ORDER BY 1, 2;
          relname         |    attname    |   atttypid   
 -------------------------+---------------+--------------
  pg_attribute            | attacl        | aclitem[]
+ pg_attribute            | attcmoptions  | text[]
  pg_attribute            | attfdwoptions | text[]
  pg_attribute            | attmissingval | anyarray
  pg_attribute            | attoptions    | text[]
@@ -107,5 +108,5 @@ ORDER BY 1, 2;
  pg_index                | indpred       | pg_node_tree
  pg_largeobject          | data          | bytea
  pg_largeobject_metadata | lomacl        | aclitem[]
-(11 rows)
+(12 rows)
 
diff --git a/src/test/regress/sql/compression.sql b/src/test/regress/sql/compression.sql
index b57f3d29ba..1e47af121b 100644
--- a/src/test/regress/sql/compression.sql
+++ b/src/test/regress/sql/compression.sql
@@ -101,6 +101,17 @@ ALTER TABLE cmdata ALTER COLUMN f1 SET COMPRESSION lz4 PRESERVE (pglz2);
 SELECT pg_column_compression(f1) FROM cmdata;
 \d+ cmdata
 
+-- compression options
+DROP TABLE cmmove1;
+CREATE TABLE cmmove1(f1 TEXT COMPRESSION pglz WITH (min_input_size '100'));
+CREATE INDEX idx1 ON cmdata(f1);
+INSERT INTO cmmove1 VALUES(repeat('1234567890',1000));
+ALTER TABLE cmmove1 ALTER COLUMN f1 SET COMPRESSION lz4 WITH (acceleration '50');
+INSERT INTO cmmove1 VALUES(repeat('1234567890',1004));
+ALTER TABLE cmmove1 ALTER COLUMN f1 SET COMPRESSION pglz WITH (min_input_size '200') PRESERVE (lz4);
+INSERT INTO cmmove1 VALUES(repeat('1234567890',1008));
+SELECT pg_column_compression(f1) FROM cmmove1;
+
 -- check data is ok
 SELECT length(f1) FROM cmdata;
 SELECT length(f1) FROM cmdata1;
-- 
2.23.0