A separate table level option to control compression

diff --git a/doc/src/sgml/ref/create_table.sgml b/doc/src/sgml/ref/create_table.sgml
index e94fe2c3b6..374f0f2579 100644
--- a/doc/src/sgml/ref/create_table.sgml
+++ b/doc/src/sgml/ref/create_table.sgml
@@ -1310,6 +1310,27 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
     </listitem>
    </varlistentry>
 
+   <varlistentry>
+    <term><literal>compress_tuple_target</literal> (<type>integer</type>)</term>
+    <listitem>
+     <para>
+     The compress_tuple_target specifies the minimum tuple length required before
+     we try to compress columns marked as Extended or Main and applies only to
+     new tuples - there is no effect on existing rows.
+     By default this parameter is set to allow at least 4 tuples per block,
+     which with the default blocksize will be 2040 bytes. Valid values are
+     between 128 bytes and the (blocksize - header), by default 8160 bytes.
+     If the value is set to a value greater than
+     <literal>toast_tuple_target</literal>, then that will be ignored and the value
+     of <literal>toast_tuple_target</literal> will be used instead.
+     Note that the default setting is often close to optimal, and
+     it is possible that setting this parameter could have negative
+     effects in some cases.
+     This parameter cannot be set for TOAST tables.
+     </para>
+    </listitem>
+   </varlistentry>
+
    <varlistentry>
     <term><literal>parallel_workers</literal> (<type>integer</type>)</term>
     <listitem>
diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index 3b0b138f24..199f1a166c 100644
--- a/src/backend/access/common/reloptions.c
+++ b/src/backend/access/common/reloptions.c
@@ -300,6 +300,15 @@ static relopt_int intRelOpts[] =
 		},
 		TOAST_TUPLE_TARGET, 128, TOAST_TUPLE_TARGET_MAIN
 	},
+	{
+		{
+			"compress_tuple_target",
+			"Sets the target tuple length at which columns will be compressed",
+			RELOPT_KIND_HEAP,
+			ShareUpdateExclusiveLock
+		},
+		TOAST_TUPLE_TARGET, 128, TOAST_TUPLE_TARGET_MAIN
+	},
 	{
 		{
 			"pages_per_range",
@@ -1377,6 +1386,8 @@ default_reloptions(Datum reloptions, bool validate, relopt_kind kind)
 		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, log_min_duration)},
 		{"toast_tuple_target", RELOPT_TYPE_INT,
 		offsetof(StdRdOptions, toast_tuple_target)},
+		{"compress_tuple_target", RELOPT_TYPE_INT,
+		offsetof(StdRdOptions, compress_tuple_target)},
 		{"autovacuum_vacuum_cost_delay", RELOPT_TYPE_REAL,
 		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, vacuum_cost_delay)},
 		{"autovacuum_vacuum_scale_factor", RELOPT_TYPE_REAL,
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 3c8a5da0bc..648559c3c8 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -2111,6 +2111,9 @@ static HeapTuple
 heap_prepare_insert(Relation relation, HeapTuple tup, TransactionId xid,
 					CommandId cid, int options)
 {
+	int		toast_tuple_threshold;
+	int		compress_tuple_threshold;
+
 	/*
 	 * Parallel operations are required to be strictly read-only in a parallel
 	 * worker.  Parallel inserts are not safe even in the leader in the
@@ -2135,6 +2138,12 @@ heap_prepare_insert(Relation relation, HeapTuple tup, TransactionId xid,
 	HeapTupleHeaderSetXmax(tup->t_data, 0); /* for cleanliness */
 	tup->t_tableOid = RelationGetRelid(relation);
 
+	toast_tuple_threshold = RelationGetToastTupleTarget(relation,
+								TOAST_TUPLE_THRESHOLD);
+	compress_tuple_threshold = RelationGetCompressTupleTarget(relation,
+								toast_tuple_threshold);
+	compress_tuple_threshold = Min(compress_tuple_threshold,
+								   toast_tuple_threshold);
 	/*
 	 * If the new tuple is too big for storage or contains already toasted
 	 * out-of-line attributes from some other relation, invoke the toaster.
@@ -2146,7 +2155,14 @@ heap_prepare_insert(Relation relation, HeapTuple tup, TransactionId xid,
 		Assert(!HeapTupleHasExternal(tup));
 		return tup;
 	}
-	else if (HeapTupleHasExternal(tup) || tup->t_len > TOAST_TUPLE_THRESHOLD)
+	/*
+	 * Activate toaster if the tuple length is greater than compression limit.
+	 * Note that compress_tuple_threshold must be less than or equal to
+	 * toast_tuple_threshold, so it's enough to only test for
+	 * compress_tuple_threshold.
+	 */
+	else if (HeapTupleHasExternal(tup) ||
+			 tup->t_len > compress_tuple_threshold)
 		return toast_insert_or_update(relation, tup, NULL, options);
 	else
 		return tup;
@@ -3464,9 +3480,26 @@ l2:
 		need_toast = false;
 	}
 	else
+	{
+		int			toast_tuple_threshold;
+		int			compress_tuple_threshold;
+
+		toast_tuple_threshold = RelationGetToastTupleTarget(relation,
+									TOAST_TUPLE_THRESHOLD);
+		compress_tuple_threshold = RelationGetCompressTupleTarget(relation,
+									toast_tuple_threshold);
+		compress_tuple_threshold = Min(compress_tuple_threshold,
+									   toast_tuple_threshold);
+
+		/*
+		 * compress_tuple_threshold must be less than or equal to
+		 * toast_tuple_threshold, so enough to only test for
+		 * compress_tuple_threshold.
+		 */
 		need_toast = (HeapTupleHasExternal(&oldtup) ||
 					  HeapTupleHasExternal(newtup) ||
-					  newtup->t_len > TOAST_TUPLE_THRESHOLD);
+					  newtup->t_len > compress_tuple_threshold);
+	}
 
 	pagefree = PageGetHeapFreeSpace(page);
 
diff --git a/src/backend/access/heap/rewriteheap.c b/src/backend/access/heap/rewriteheap.c
index bce4274362..9b7fe81235 100644
--- a/src/backend/access/heap/rewriteheap.c
+++ b/src/backend/access/heap/rewriteheap.c
@@ -636,7 +636,15 @@ raw_heap_insert(RewriteState state, HeapTuple tup)
 	Size		len;
 	OffsetNumber newoff;
 	HeapTuple	heaptup;
-
+	int			toast_tuple_threshold;
+	int			compress_tuple_threshold;
+
+	toast_tuple_threshold = RelationGetToastTupleTarget(state->rs_new_rel,
+									TOAST_TUPLE_THRESHOLD);
+	compress_tuple_threshold = RelationGetCompressTupleTarget(state->rs_new_rel,
+									toast_tuple_threshold);
+	compress_tuple_threshold = Min(compress_tuple_threshold,
+								   toast_tuple_threshold);
 	/*
 	 * If the new tuple is too big for storage or contains already toasted
 	 * out-of-line attributes from some other relation, invoke the toaster.
@@ -650,7 +658,14 @@ raw_heap_insert(RewriteState state, HeapTuple tup)
 		Assert(!HeapTupleHasExternal(tup));
 		heaptup = tup;
 	}
-	else if (HeapTupleHasExternal(tup) || tup->t_len > TOAST_TUPLE_THRESHOLD)
+	/*
+	 * Activate toaster if the tuple length is greater than compression limit.
+	 * Note that compress_tuple_threshold must be less than or equal to
+	 * toast_tuple_threshold, so it's enough to only test for
+	 * compress_tuple_threshold.
+	 */
+	else if (HeapTupleHasExternal(tup) ||
+			 tup->t_len > compress_tuple_threshold)
 	{
 		int options = HEAP_INSERT_SKIP_FSM;
 
diff --git a/src/backend/access/heap/tuptoaster.c b/src/backend/access/heap/tuptoaster.c
index cd921a4600..ac0444d05e 100644
--- a/src/backend/access/heap/tuptoaster.c
+++ b/src/backend/access/heap/tuptoaster.c
@@ -544,6 +544,7 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
 	bool		has_nulls = false;
 
 	Size		maxDataLen;
+	Size		maxCompressLen;
 	Size		hoff;
 
 	char		toast_action[MaxHeapAttributeNumber];
@@ -726,12 +727,19 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
 	/* now convert to a limit on the tuple data size */
 	maxDataLen = RelationGetToastTupleTarget(rel, TOAST_TUPLE_TARGET) - hoff;
 
+	/*
+	 * Get the limit at which we should apply compression. This will be same as
+	 * maxDataLen unless overridden by the user explicitly.
+	 */
+	maxCompressLen = RelationGetCompressTupleTarget(rel, maxDataLen) - hoff;
+	maxCompressLen = Min(maxCompressLen, maxDataLen);
+
 	/*
 	 * Look for attributes with attstorage 'x' to compress.  Also find large
 	 * attributes with attstorage 'x' or 'e', and store them external.
 	 */
 	while (heap_compute_data_size(tupleDesc,
-								  toast_values, toast_isnull) > maxDataLen)
+								  toast_values, toast_isnull) > maxCompressLen)
 	{
 		int			biggest_attno = -1;
 		int32		biggest_size = MAXALIGN(TOAST_POINTER_SIZE);
@@ -876,7 +884,7 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
 	 * compression
 	 */
 	while (heap_compute_data_size(tupleDesc,
-								  toast_values, toast_isnull) > maxDataLen)
+								  toast_values, toast_isnull) > maxCompressLen)
 	{
 		int			biggest_attno = -1;
 		int32		biggest_size = MAXALIGN(TOAST_POINTER_SIZE);
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index 54028515a7..9e17e2a599 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -263,6 +263,7 @@ typedef struct StdRdOptions
 	/* fraction of newly inserted tuples prior to trigger index cleanup */
 	float8		vacuum_cleanup_index_scale_factor;
 	int			toast_tuple_target; /* target for tuple toasting */
+	int			compress_tuple_target; /* target for tuple compression */
 	AutoVacOpts autovacuum;		/* autovacuum-related options */
 	bool		user_catalog_table; /* use as an additional catalog relation */
 	int			parallel_workers;	/* max number of parallel workers */
@@ -279,6 +280,14 @@ typedef struct StdRdOptions
 	((relation)->rd_options ? \
 	 ((StdRdOptions *) (relation)->rd_options)->toast_tuple_target : (defaulttarg))
 
+/*
+ * RelationGetCompressTupleTarget
+ *		Returns the relation's compress_tuple_target.  Note multiple eval of argument!
+ */
+#define RelationGetCompressTupleTarget(relation, defaulttarg) \
+	((relation)->rd_options ? \
+	 ((StdRdOptions *) (relation)->rd_options)->compress_tuple_target : (defaulttarg))
+
 /*
  * RelationGetFillFactor
  *		Returns the relation's fillfactor.  Note multiple eval of argument!
diff --git a/src/test/regress/expected/strings.out b/src/test/regress/expected/strings.out
index 189bdffdca..2f1a364b3c 100644
--- a/src/test/regress/expected/strings.out
+++ b/src/test/regress/expected/strings.out
@@ -1209,7 +1209,70 @@ select 0 = pg_relation_size('pg_toast.pg_toast_'||(select oid from pg_class wher
  t
 (1 row)
 
+CREATE TABLE compresstest250 (a int, b text) WITH (compress_tuple_target = 250);
+CREATE TABLE compresstest2040 (a int, b text) WITH (compress_tuple_target = 2040);
+-- shouldn't get compressed nor toasted
+INSERT INTO compresstest250 VALUES (1, repeat('1234567890',20));
+-- should get compressed, but not toasted
+INSERT INTO compresstest250 VALUES (2, repeat('1234567890',30));
+-- shouldn't get compressed nor toasted
+INSERT INTO compresstest2040 VALUES (1, repeat('1234567890',20));
+-- shouldn't get compressed nor toasted
+INSERT INTO compresstest2040 VALUES (2, repeat('1234567890',30));
+SELECT a, pg_column_size(b) FROM compresstest250;
+ a | pg_column_size 
+---+----------------
+ 1 |            204
+ 2 |             25
+(2 rows)
+
+SELECT a, pg_column_size(b) FROM compresstest2040;
+ a | pg_column_size 
+---+----------------
+ 1 |            204
+ 2 |            304
+(2 rows)
+
+-- expect 0 blocks
+select 0 = pg_relation_size('pg_toast.pg_toast_'||(select oid from pg_class where relname = 'compresstest250'))/current_setting('block_size')::integer as blocks;
+ blocks 
+--------
+ t
+(1 row)
+
+select 0 = pg_relation_size('pg_toast.pg_toast_'||(select oid from pg_class where relname = 'compresstest2040'))/current_setting('block_size')::integer as blocks;
+ blocks 
+--------
+ t
+(1 row)
+
+-- should get compressed, and toasted
+INSERT INTO compresstest250 SELECT 3, string_agg('', md5(random()::text)) FROM generate_series(1,300);
+-- expect > 0 blocks
+select 0 = pg_relation_size('pg_toast.pg_toast_'||(select oid from pg_class where relname = 'compresstest250'))/current_setting('block_size')::integer as blocks;
+ blocks 
+--------
+ f
+(1 row)
+
+ALTER TABLE compresstest250 RENAME TO compresstest1500;
+ALTER TABLE compresstest1500 SET (compress_tuple_target=1500);
+INSERT INTO compresstest1500 VALUES (4, repeat('1234567890',140));
+INSERT INTO compresstest2040 VALUES (4, repeat('1234567890',140));
+SELECT a, pg_column_size(b) FROM compresstest1500 WHERE a = 4;
+ a | pg_column_size 
+---+----------------
+ 4 |           1404
+(1 row)
+
+SELECT a, pg_column_size(b) FROM compresstest2040 WHERE a = 4;
+ a | pg_column_size 
+---+----------------
+ 4 |           1404
+(1 row)
+
 DROP TABLE toasttest;
+DROP TABLE compresstest1500, compresstest2040;
 --
 -- test substr with toasted bytea values
 --
diff --git a/src/test/regress/sql/strings.sql b/src/test/regress/sql/strings.sql
index f2203ef1b1..1fb04a3a1b 100644
--- a/src/test/regress/sql/strings.sql
+++ b/src/test/regress/sql/strings.sql
@@ -386,7 +386,38 @@ INSERT INTO toasttest values (repeat('1234567890',300));
 -- expect 0 blocks
 select 0 = pg_relation_size('pg_toast.pg_toast_'||(select oid from pg_class where relname = 'toasttest'))/current_setting('block_size')::integer as blocks;
 
+CREATE TABLE compresstest250 (a int, b text) WITH (compress_tuple_target = 250);
+CREATE TABLE compresstest2040 (a int, b text) WITH (compress_tuple_target = 2040);
+-- shouldn't get compressed nor toasted
+INSERT INTO compresstest250 VALUES (1, repeat('1234567890',20));
+-- should get compressed, but not toasted
+INSERT INTO compresstest250 VALUES (2, repeat('1234567890',30));
+-- shouldn't get compressed nor toasted
+INSERT INTO compresstest2040 VALUES (1, repeat('1234567890',20));
+-- shouldn't get compressed nor toasted
+INSERT INTO compresstest2040 VALUES (2, repeat('1234567890',30));
+
+SELECT a, pg_column_size(b) FROM compresstest250;
+SELECT a, pg_column_size(b) FROM compresstest2040;
+-- expect 0 blocks
+select 0 = pg_relation_size('pg_toast.pg_toast_'||(select oid from pg_class where relname = 'compresstest250'))/current_setting('block_size')::integer as blocks;
+select 0 = pg_relation_size('pg_toast.pg_toast_'||(select oid from pg_class where relname = 'compresstest2040'))/current_setting('block_size')::integer as blocks;
+
+-- should get compressed, and toasted
+INSERT INTO compresstest250 SELECT 3, string_agg('', md5(random()::text)) FROM generate_series(1,300);
+-- expect > 0 blocks
+select 0 = pg_relation_size('pg_toast.pg_toast_'||(select oid from pg_class where relname = 'compresstest250'))/current_setting('block_size')::integer as blocks;
+
+ALTER TABLE compresstest250 RENAME TO compresstest1500;
+ALTER TABLE compresstest1500 SET (compress_tuple_target=1500);
+INSERT INTO compresstest1500 VALUES (4, repeat('1234567890',140));
+INSERT INTO compresstest2040 VALUES (4, repeat('1234567890',140));
+
+SELECT a, pg_column_size(b) FROM compresstest1500 WHERE a = 4;
+SELECT a, pg_column_size(b) FROM compresstest2040 WHERE a = 4;
+
 DROP TABLE toasttest;
+DROP TABLE compresstest1500, compresstest2040;
 
 --
 -- test substr with toasted bytea values

From d8672d9f11351cd97b940a81784627ad0bf07d7e Mon Sep 17 00:00:00 2001
From: Pavan Deolasee <pavan.deolasee@gmail.com>
Date: Wed, 6 Feb 2019 16:37:48 +0530
Subject: [PATCH] Add a table level option to control compression

We now support a new table level option, compress_tuple_target, which can be
set to decide when to trigger compression on a tuple. Prior to this, the
toast_tuple_target (or the compile time default) was used to decide whether or
not to compress a tuple. But this was detrimental when a tuple is large enough
and has a good compression ratio, but not large enough to cross the toast
threshold. We wouldn't compress such tuples, thus wasting a potentially good
opportunity.
---
 doc/src/sgml/ref/create_table.sgml     | 27 +++++++++++++++
 doc/src/sgml/storage.sgml              | 29 +++++++++++-----
 src/backend/access/common/reloptions.c | 11 ++++++
 src/backend/access/heap/heapam.c       | 37 ++++++++++++++++++--
 src/backend/access/heap/rewriteheap.c  | 19 ++++++++--
 src/backend/access/heap/tuptoaster.c   | 12 +++++--
 src/include/utils/rel.h                |  9 +++++
 src/test/regress/expected/strings.out  | 63 ++++++++++++++++++++++++++++++++++
 src/test/regress/sql/strings.sql       | 31 +++++++++++++++++
 9 files changed, 224 insertions(+), 14 deletions(-)

diff --git a/doc/src/sgml/ref/create_table.sgml b/doc/src/sgml/ref/create_table.sgml
index e94fe2c3b6..18cd06dd66 100644
--- a/doc/src/sgml/ref/create_table.sgml
+++ b/doc/src/sgml/ref/create_table.sgml
@@ -1310,6 +1310,33 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
     </listitem>
    </varlistentry>
 
+   <varlistentry>
+    <term><literal>compress_tuple_target</literal> (<type>integer</type>)</term>
+    <listitem>
+     <para>
+     The compress_tuple_target parameter specifies the minimum tuple length required before
+     we try to compress columns marked as Extended or Main and applies only to
+     new tuples - there is no effect on existing rows.
+     By default this parameter is set to allow at least 4 tuples per block,
+     which with the default blocksize will be 2040 bytes. Valid values are
+     between 128 bytes and (blocksize - header), by default 8160 bytes.
+     If the specified value is greater than
+     <literal>toast_tuple_target</literal>, then
+     we will use the current setting of <literal>toast_tuple_target</literal>
+     for <literal>compress_tuple_target</literal>.
+     Note that the default setting is often close to optimal. If the value is
+     set too low then the <acronym>TOAST</acronym> may get invoked too often.
+     If the compressibility of
+     the field values is not good, then compression and decompression can add
+     significant computation overhead without corresponding savings in storage
+     consumption.
+    </para>
+    <para>
+     This parameter cannot be set for TOAST tables.
+     </para>
+    </listitem>
+   </varlistentry>
+
    <varlistentry>
     <term><literal>parallel_workers</literal> (<type>integer</type>)</term>
     <listitem>
diff --git a/doc/src/sgml/storage.sgml b/doc/src/sgml/storage.sgml
index cbdad0c3fb..e65ae5ff88 100644
--- a/doc/src/sgml/storage.sgml
+++ b/doc/src/sgml/storage.sgml
@@ -425,9 +425,13 @@ bytes regardless of the actual size of the represented value.
 
 <para>
 The <acronym>TOAST</acronym> management code is triggered only
-when a row value to be stored in a table is wider than
-<symbol>TOAST_TUPLE_THRESHOLD</symbol> bytes (normally 2 kB).
-The <acronym>TOAST</acronym> code will compress and/or move
+when a row value to be stored in a table is wider than compile time default of
+<symbol>TOAST_TUPLE_THRESHOLD</symbol> bytes (normally 2 kB) or a table level
+<symbol>COMPRESS_TUPLE_TARGET</symbol> option. 
+The <acronym>TOAST</acronym> code will first in-line compress field
+values greater than <symbol>COMPRESS_TUPLE_TARGET</symbol>. If the compressed
+row value is still greater than <symbol>TOAST_TUPLE_TARGET</symbol> then the
+<acronym>TOAST</acronym> code will move the
 field values out-of-line until the row value is shorter than
 <symbol>TOAST_TUPLE_TARGET</symbol> bytes (also normally 2 kB, adjustable)
 or no more gains can be had.  During an UPDATE
@@ -436,6 +440,20 @@ UPDATE of a row with out-of-line values incurs no <acronym>TOAST</acronym> costs
 none of the out-of-line values change.
 </para>
 
+<para>
+<symbol>TOAST_TUPLE_TARGET</symbol> can be adjusted for each table using
+<link linkend="sql-altertable"><command>ALTER TABLE ... SET (toast_tuple_target = N)</command></link>
+</para>
+
+<para>
+<symbol>COMPRESS_TUPLE_TARGET</symbol> can be adjusted for each table using
+<link linkend="sql-altertable"><command>ALTER TABLE ... SET
+(compress_tuple_target = N)</command></link>. If
+<symbol>COMPRESS_TUPLE_TARGET</symbol> is not set explicitly for a table or if
+it is greater than <symbol>TOAST_TUPLE_TARGET</symbol> then
+<symbol>TOAST_TUPLE_TARGET</symbol> value is used to decide when to compress.
+</para>
+
 <para>
 The <acronym>TOAST</acronym> management code recognizes four different strategies
 for storing <acronym>TOAST</acronym>-able columns on disk:
@@ -483,11 +501,6 @@ of that data type, but the strategy for a given table column can be altered
 with <link linkend="sql-altertable"><command>ALTER TABLE ... SET STORAGE</command></link>.
 </para>
 
-<para>
-<symbol>TOAST_TUPLE_TARGET</symbol> can be adjusted for each table using
-<link linkend="sql-altertable"><command>ALTER TABLE ... SET (toast_tuple_target = N)</command></link>
-</para>
-
 <para>
 This scheme has a number of advantages compared to a more straightforward
 approach such as allowing row values to span pages.  Assuming that queries are
diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index 3b0b138f24..199f1a166c 100644
--- a/src/backend/access/common/reloptions.c
+++ b/src/backend/access/common/reloptions.c
@@ -300,6 +300,15 @@ static relopt_int intRelOpts[] =
 		},
 		TOAST_TUPLE_TARGET, 128, TOAST_TUPLE_TARGET_MAIN
 	},
+	{
+		{
+			"compress_tuple_target",
+			"Sets the target tuple length at which columns will be compressed",
+			RELOPT_KIND_HEAP,
+			ShareUpdateExclusiveLock
+		},
+		TOAST_TUPLE_TARGET, 128, TOAST_TUPLE_TARGET_MAIN
+	},
 	{
 		{
 			"pages_per_range",
@@ -1377,6 +1386,8 @@ default_reloptions(Datum reloptions, bool validate, relopt_kind kind)
 		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, log_min_duration)},
 		{"toast_tuple_target", RELOPT_TYPE_INT,
 		offsetof(StdRdOptions, toast_tuple_target)},
+		{"compress_tuple_target", RELOPT_TYPE_INT,
+		offsetof(StdRdOptions, compress_tuple_target)},
 		{"autovacuum_vacuum_cost_delay", RELOPT_TYPE_REAL,
 		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, vacuum_cost_delay)},
 		{"autovacuum_vacuum_scale_factor", RELOPT_TYPE_REAL,
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 3c8a5da0bc..648559c3c8 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -2111,6 +2111,9 @@ static HeapTuple
 heap_prepare_insert(Relation relation, HeapTuple tup, TransactionId xid,
 					CommandId cid, int options)
 {
+	int		toast_tuple_threshold;
+	int		compress_tuple_threshold;
+
 	/*
 	 * Parallel operations are required to be strictly read-only in a parallel
 	 * worker.  Parallel inserts are not safe even in the leader in the
@@ -2135,6 +2138,12 @@ heap_prepare_insert(Relation relation, HeapTuple tup, TransactionId xid,
 	HeapTupleHeaderSetXmax(tup->t_data, 0); /* for cleanliness */
 	tup->t_tableOid = RelationGetRelid(relation);
 
+	toast_tuple_threshold = RelationGetToastTupleTarget(relation,
+								TOAST_TUPLE_THRESHOLD);
+	compress_tuple_threshold = RelationGetCompressTupleTarget(relation,
+								toast_tuple_threshold);
+	compress_tuple_threshold = Min(compress_tuple_threshold,
+								   toast_tuple_threshold);
 	/*
 	 * If the new tuple is too big for storage or contains already toasted
 	 * out-of-line attributes from some other relation, invoke the toaster.
@@ -2146,7 +2155,14 @@ heap_prepare_insert(Relation relation, HeapTuple tup, TransactionId xid,
 		Assert(!HeapTupleHasExternal(tup));
 		return tup;
 	}
-	else if (HeapTupleHasExternal(tup) || tup->t_len > TOAST_TUPLE_THRESHOLD)
+	/*
+	 * Activate toaster if the tuple length is greater than compression limit.
+	 * Note that compress_tuple_threshold must be less than or equal to
+	 * toast_tuple_threshold, so it's enough to only test for
+	 * compress_tuple_threshold.
+	 */
+	else if (HeapTupleHasExternal(tup) ||
+			 tup->t_len > compress_tuple_threshold)
 		return toast_insert_or_update(relation, tup, NULL, options);
 	else
 		return tup;
@@ -3464,9 +3480,26 @@ l2:
 		need_toast = false;
 	}
 	else
+	{
+		int			toast_tuple_threshold;
+		int			compress_tuple_threshold;
+
+		toast_tuple_threshold = RelationGetToastTupleTarget(relation,
+									TOAST_TUPLE_THRESHOLD);
+		compress_tuple_threshold = RelationGetCompressTupleTarget(relation,
+									toast_tuple_threshold);
+		compress_tuple_threshold = Min(compress_tuple_threshold,
+									   toast_tuple_threshold);
+
+		/*
+		 * compress_tuple_threshold must be less than or equal to
+		 * toast_tuple_threshold, so enough to only test for
+		 * compress_tuple_threshold.
+		 */
 		need_toast = (HeapTupleHasExternal(&oldtup) ||
 					  HeapTupleHasExternal(newtup) ||
-					  newtup->t_len > TOAST_TUPLE_THRESHOLD);
+					  newtup->t_len > compress_tuple_threshold);
+	}
 
 	pagefree = PageGetHeapFreeSpace(page);
 
diff --git a/src/backend/access/heap/rewriteheap.c b/src/backend/access/heap/rewriteheap.c
index bce4274362..9b7fe81235 100644
--- a/src/backend/access/heap/rewriteheap.c
+++ b/src/backend/access/heap/rewriteheap.c
@@ -636,7 +636,15 @@ raw_heap_insert(RewriteState state, HeapTuple tup)
 	Size		len;
 	OffsetNumber newoff;
 	HeapTuple	heaptup;
-
+	int			toast_tuple_threshold;
+	int			compress_tuple_threshold;
+
+	toast_tuple_threshold = RelationGetToastTupleTarget(state->rs_new_rel,
+									TOAST_TUPLE_THRESHOLD);
+	compress_tuple_threshold = RelationGetCompressTupleTarget(state->rs_new_rel,
+									toast_tuple_threshold);
+	compress_tuple_threshold = Min(compress_tuple_threshold,
+								   toast_tuple_threshold);
 	/*
 	 * If the new tuple is too big for storage or contains already toasted
 	 * out-of-line attributes from some other relation, invoke the toaster.
@@ -650,7 +658,14 @@ raw_heap_insert(RewriteState state, HeapTuple tup)
 		Assert(!HeapTupleHasExternal(tup));
 		heaptup = tup;
 	}
-	else if (HeapTupleHasExternal(tup) || tup->t_len > TOAST_TUPLE_THRESHOLD)
+	/*
+	 * Activate toaster if the tuple length is greater than compression limit.
+	 * Note that compress_tuple_threshold must be less than or equal to
+	 * toast_tuple_threshold, so it's enough to only test for
+	 * compress_tuple_threshold.
+	 */
+	else if (HeapTupleHasExternal(tup) ||
+			 tup->t_len > compress_tuple_threshold)
 	{
 		int options = HEAP_INSERT_SKIP_FSM;
 
diff --git a/src/backend/access/heap/tuptoaster.c b/src/backend/access/heap/tuptoaster.c
index cd921a4600..ac0444d05e 100644
--- a/src/backend/access/heap/tuptoaster.c
+++ b/src/backend/access/heap/tuptoaster.c
@@ -544,6 +544,7 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
 	bool		has_nulls = false;
 
 	Size		maxDataLen;
+	Size		maxCompressLen;
 	Size		hoff;
 
 	char		toast_action[MaxHeapAttributeNumber];
@@ -726,12 +727,19 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
 	/* now convert to a limit on the tuple data size */
 	maxDataLen = RelationGetToastTupleTarget(rel, TOAST_TUPLE_TARGET) - hoff;
 
+	/*
+	 * Get the limit at which we should apply compression. This will be same as
+	 * maxDataLen unless overridden by the user explicitly.
+	 */
+	maxCompressLen = RelationGetCompressTupleTarget(rel, maxDataLen) - hoff;
+	maxCompressLen = Min(maxCompressLen, maxDataLen);
+
 	/*
 	 * Look for attributes with attstorage 'x' to compress.  Also find large
 	 * attributes with attstorage 'x' or 'e', and store them external.
 	 */
 	while (heap_compute_data_size(tupleDesc,
-								  toast_values, toast_isnull) > maxDataLen)
+								  toast_values, toast_isnull) > maxCompressLen)
 	{
 		int			biggest_attno = -1;
 		int32		biggest_size = MAXALIGN(TOAST_POINTER_SIZE);
@@ -876,7 +884,7 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
 	 * compression
 	 */
 	while (heap_compute_data_size(tupleDesc,
-								  toast_values, toast_isnull) > maxDataLen)
+								  toast_values, toast_isnull) > maxCompressLen)
 	{
 		int			biggest_attno = -1;
 		int32		biggest_size = MAXALIGN(TOAST_POINTER_SIZE);
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index 54028515a7..9e17e2a599 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -263,6 +263,7 @@ typedef struct StdRdOptions
 	/* fraction of newly inserted tuples prior to trigger index cleanup */
 	float8		vacuum_cleanup_index_scale_factor;
 	int			toast_tuple_target; /* target for tuple toasting */
+	int			compress_tuple_target; /* target for tuple compression */
 	AutoVacOpts autovacuum;		/* autovacuum-related options */
 	bool		user_catalog_table; /* use as an additional catalog relation */
 	int			parallel_workers;	/* max number of parallel workers */
@@ -279,6 +280,14 @@ typedef struct StdRdOptions
 	((relation)->rd_options ? \
 	 ((StdRdOptions *) (relation)->rd_options)->toast_tuple_target : (defaulttarg))
 
+/*
+ * RelationGetCompressTupleTarget
+ *		Returns the relation's compress_tuple_target.  Note multiple eval of argument!
+ */
+#define RelationGetCompressTupleTarget(relation, defaulttarg) \
+	((relation)->rd_options ? \
+	 ((StdRdOptions *) (relation)->rd_options)->compress_tuple_target : (defaulttarg))
+
 /*
  * RelationGetFillFactor
  *		Returns the relation's fillfactor.  Note multiple eval of argument!
diff --git a/src/test/regress/expected/strings.out b/src/test/regress/expected/strings.out
index 189bdffdca..2f1a364b3c 100644
--- a/src/test/regress/expected/strings.out
+++ b/src/test/regress/expected/strings.out
@@ -1209,7 +1209,70 @@ select 0 = pg_relation_size('pg_toast.pg_toast_'||(select oid from pg_class wher
  t
 (1 row)
 
+CREATE TABLE compresstest250 (a int, b text) WITH (compress_tuple_target = 250);
+CREATE TABLE compresstest2040 (a int, b text) WITH (compress_tuple_target = 2040);
+-- shouldn't get compressed nor toasted
+INSERT INTO compresstest250 VALUES (1, repeat('1234567890',20));
+-- should get compressed, but not toasted
+INSERT INTO compresstest250 VALUES (2, repeat('1234567890',30));
+-- shouldn't get compressed nor toasted
+INSERT INTO compresstest2040 VALUES (1, repeat('1234567890',20));
+-- shouldn't get compressed nor toasted
+INSERT INTO compresstest2040 VALUES (2, repeat('1234567890',30));
+SELECT a, pg_column_size(b) FROM compresstest250;
+ a | pg_column_size 
+---+----------------
+ 1 |            204
+ 2 |             25
+(2 rows)
+
+SELECT a, pg_column_size(b) FROM compresstest2040;
+ a | pg_column_size 
+---+----------------
+ 1 |            204
+ 2 |            304
+(2 rows)
+
+-- expect 0 blocks
+select 0 = pg_relation_size('pg_toast.pg_toast_'||(select oid from pg_class where relname = 'compresstest250'))/current_setting('block_size')::integer as blocks;
+ blocks 
+--------
+ t
+(1 row)
+
+select 0 = pg_relation_size('pg_toast.pg_toast_'||(select oid from pg_class where relname = 'compresstest2040'))/current_setting('block_size')::integer as blocks;
+ blocks 
+--------
+ t
+(1 row)
+
+-- should get compressed, and toasted
+INSERT INTO compresstest250 SELECT 3, string_agg('', md5(random()::text)) FROM generate_series(1,300);
+-- expect > 0 blocks
+select 0 = pg_relation_size('pg_toast.pg_toast_'||(select oid from pg_class where relname = 'compresstest250'))/current_setting('block_size')::integer as blocks;
+ blocks 
+--------
+ f
+(1 row)
+
+ALTER TABLE compresstest250 RENAME TO compresstest1500;
+ALTER TABLE compresstest1500 SET (compress_tuple_target=1500);
+INSERT INTO compresstest1500 VALUES (4, repeat('1234567890',140));
+INSERT INTO compresstest2040 VALUES (4, repeat('1234567890',140));
+SELECT a, pg_column_size(b) FROM compresstest1500 WHERE a = 4;
+ a | pg_column_size 
+---+----------------
+ 4 |           1404
+(1 row)
+
+SELECT a, pg_column_size(b) FROM compresstest2040 WHERE a = 4;
+ a | pg_column_size 
+---+----------------
+ 4 |           1404
+(1 row)
+
 DROP TABLE toasttest;
+DROP TABLE compresstest1500, compresstest2040;
 --
 -- test substr with toasted bytea values
 --
diff --git a/src/test/regress/sql/strings.sql b/src/test/regress/sql/strings.sql
index f2203ef1b1..1fb04a3a1b 100644
--- a/src/test/regress/sql/strings.sql
+++ b/src/test/regress/sql/strings.sql
@@ -386,7 +386,38 @@ INSERT INTO toasttest values (repeat('1234567890',300));
 -- expect 0 blocks
 select 0 = pg_relation_size('pg_toast.pg_toast_'||(select oid from pg_class where relname = 'toasttest'))/current_setting('block_size')::integer as blocks;
 
+CREATE TABLE compresstest250 (a int, b text) WITH (compress_tuple_target = 250);
+CREATE TABLE compresstest2040 (a int, b text) WITH (compress_tuple_target = 2040);
+-- shouldn't get compressed nor toasted
+INSERT INTO compresstest250 VALUES (1, repeat('1234567890',20));
+-- should get compressed, but not toasted
+INSERT INTO compresstest250 VALUES (2, repeat('1234567890',30));
+-- shouldn't get compressed nor toasted
+INSERT INTO compresstest2040 VALUES (1, repeat('1234567890',20));
+-- shouldn't get compressed nor toasted
+INSERT INTO compresstest2040 VALUES (2, repeat('1234567890',30));
+
+SELECT a, pg_column_size(b) FROM compresstest250;
+SELECT a, pg_column_size(b) FROM compresstest2040;
+-- expect 0 blocks
+select 0 = pg_relation_size('pg_toast.pg_toast_'||(select oid from pg_class where relname = 'compresstest250'))/current_setting('block_size')::integer as blocks;
+select 0 = pg_relation_size('pg_toast.pg_toast_'||(select oid from pg_class where relname = 'compresstest2040'))/current_setting('block_size')::integer as blocks;
+
+-- should get compressed, and toasted
+INSERT INTO compresstest250 SELECT 3, string_agg('', md5(random()::text)) FROM generate_series(1,300);
+-- expect > 0 blocks
+select 0 = pg_relation_size('pg_toast.pg_toast_'||(select oid from pg_class where relname = 'compresstest250'))/current_setting('block_size')::integer as blocks;
+
+ALTER TABLE compresstest250 RENAME TO compresstest1500;
+ALTER TABLE compresstest1500 SET (compress_tuple_target=1500);
+INSERT INTO compresstest1500 VALUES (4, repeat('1234567890',140));
+INSERT INTO compresstest2040 VALUES (4, repeat('1234567890',140));
+
+SELECT a, pg_column_size(b) FROM compresstest1500 WHERE a = 4;
+SELECT a, pg_column_size(b) FROM compresstest2040 WHERE a = 4;
+
 DROP TABLE toasttest;
+DROP TABLE compresstest1500, compresstest2040;
 
 --
 -- test substr with toasted bytea values
-- 
2.14.3 (Apple Git-98)

diff --git a/doc/src/sgml/ref/create_table.sgml b/doc/src/sgml/ref/create_table.sgml
index 0fcbc660b3..acb858fba1 100644
--- a/doc/src/sgml/ref/create_table.sgml
+++ b/doc/src/sgml/ref/create_table.sgml
@@ -1343,6 +1343,32 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
     </listitem>
    </varlistentry>
 
+   <varlistentry>
+    <term><literal>compress_tuple_target</literal> (<type>integer</type>)</term>
+    <listitem>
+     <para>
+      The compress_tuple_target parameter specifies the minimum tuple length
+      required before we try to compress columns marked as Extended or Main
+      and applies only to new tuples - there is no effect on existing rows.
+      By default this parameter is set to allow at least 4 tuples per block,
+      which with the default blocksize will be 2040 bytes. Valid values are
+      between 128 bytes and (blocksize - header), by default 8160 bytes.
+      If the specified value is greater than
+      <literal>toast_tuple_target</literal>, then we will use the current
+      setting of <literal>toast_tuple_target</literal> for
+      <literal>compress_tuple_target</literal>. Note that the default setting
+      is often close to optimal. If the value is set too low then the
+      <acronym>TOAST</acronym> may get invoked too often. If the
+      compressibility of the field values is not good, then compression and
+      decompression can add significant computation overhead without
+      corresponding savings in storage consumption.
+     </para>
+     <para>
+      This parameter cannot be set for TOAST tables.
+     </para>
+    </listitem>
+   </varlistentry>
+
    <varlistentry>
     <term><literal>parallel_workers</literal> (<type>integer</type>)</term>
     <listitem>
diff --git a/doc/src/sgml/storage.sgml b/doc/src/sgml/storage.sgml
index 5df987f9c9..cfa0af571d 100644
--- a/doc/src/sgml/storage.sgml
+++ b/doc/src/sgml/storage.sgml
@@ -426,14 +426,17 @@ bytes regardless of the actual size of the represented value.
 <para>
 The <acronym>TOAST</acronym> management code is triggered only
 when a row value to be stored in a table is wider than
-<symbol>TOAST_TUPLE_THRESHOLD</symbol> bytes (normally 2 kB).
-The <acronym>TOAST</acronym> code will compress and/or move
-field values out-of-line until the row value is shorter than
-<symbol>TOAST_TUPLE_TARGET</symbol> bytes (also normally 2 kB, adjustable)
-or no more gains can be had.  During an UPDATE
+<symbol>TOAST_TUPLE_THRESHOLD</symbol> bytes (normally 2 kB) or the
+table-level <symbol>COMPRESS_TUPLE_TARGET</symbol> option.  The
+<acronym>TOAST</acronym> code will first in-line compress field values
+greater than <symbol>COMPRESS_TUPLE_TARGET</symbol>. If the compressed row
+value is still greater than <symbol>TOAST_TUPLE_TARGET</symbol> then the
+<acronym>TOAST</acronym> code will move the field values out-of-line until
+the row value is shorter than <symbol>TOAST_TUPLE_TARGET</symbol> bytes (also
+normally 2 kB, adjustable) or no more gains can be had.  During an UPDATE
 operation, values of unchanged fields are normally preserved as-is; so an
-UPDATE of a row with out-of-line values incurs no <acronym>TOAST</acronym> costs if
-none of the out-of-line values change.
+UPDATE of a row with out-of-line values incurs no <acronym>TOAST</acronym>
+costs if none of the out-of-line values change.
 </para>
 
 <para>
@@ -485,7 +488,17 @@ with <link linkend="sql-altertable"><command>ALTER TABLE ... SET STORAGE</comman
 
 <para>
 <symbol>TOAST_TUPLE_TARGET</symbol> can be adjusted for each table using
-<link linkend="sql-altertable"><command>ALTER TABLE ... SET (toast_tuple_target = N)</command></link>
+<link linkend="sql-altertable"><command>ALTER TABLE ... SET (toast_tuple_target = N)</command></link>.
+</para>
+
+<para>
+<symbol>COMPRESS_TUPLE_TARGET</symbol> can be adjusted for each table using
+<link linkend="sql-altertable"><command>ALTER TABLE ... SET
+(compress_tuple_target = N)</command></link>. If
+<symbol>COMPRESS_TUPLE_TARGET</symbol> is not set explicitly for a table or if
+it is greater than <symbol>TOAST_TUPLE_TARGET</symbol> then
+<symbol>TOAST_TUPLE_TARGET</symbol> value is used to decide when to do
+compression.
 </para>
 
 <para>
diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index b58a1f7a72..479f961a49 100644
--- a/src/backend/access/common/reloptions.c
+++ b/src/backend/access/common/reloptions.c
@@ -291,6 +291,15 @@ static relopt_int intRelOpts[] =
 		},
 		TOAST_TUPLE_TARGET, 128, TOAST_TUPLE_TARGET_MAIN
 	},
+	{
+		{
+			"compress_tuple_target",
+			"Sets the target tuple length at which columns will be compressed",
+			RELOPT_KIND_HEAP,
+			ShareUpdateExclusiveLock
+		},
+		COMPRESS_TUPLE_TARGET, 128, TOAST_TUPLE_TARGET_MAIN
+	},
 	{
 		{
 			"pages_per_range",
@@ -1377,6 +1386,8 @@ default_reloptions(Datum reloptions, bool validate, relopt_kind kind)
 		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, log_min_duration)},
 		{"toast_tuple_target", RELOPT_TYPE_INT,
 		offsetof(StdRdOptions, toast_tuple_target)},
+		{"compress_tuple_target", RELOPT_TYPE_INT,
+		offsetof(StdRdOptions, compress_tuple_target)},
 		{"autovacuum_vacuum_cost_delay", RELOPT_TYPE_REAL,
 		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, vacuum_cost_delay)},
 		{"autovacuum_vacuum_scale_factor", RELOPT_TYPE_REAL,
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 05ceb6550d..01e8224fd4 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -2053,6 +2053,9 @@ static HeapTuple
 heap_prepare_insert(Relation relation, HeapTuple tup, TransactionId xid,
 					CommandId cid, int options)
 {
+	int		toast_tuple_threshold;
+	int		compress_tuple_threshold;
+
 	/*
 	 * Parallel operations are required to be strictly read-only in a parallel
 	 * worker.  Parallel inserts are not safe even in the leader in the
@@ -2077,9 +2080,25 @@ heap_prepare_insert(Relation relation, HeapTuple tup, TransactionId xid,
 	HeapTupleHeaderSetXmax(tup->t_data, 0); /* for cleanliness */
 	tup->t_tableOid = RelationGetRelid(relation);
 
+	/*
+	 * Compute the compressibility threshold, which is the minimum between
+	 * toast_tuple_target and compress_tuple_target for a relation.
+	 */
+	toast_tuple_threshold = RelationGetToastTupleTarget(relation,
+								TOAST_TUPLE_TARGET);
+	compress_tuple_threshold = RelationGetCompressTupleTarget(relation,
+								COMPRESS_TUPLE_TARGET);
+	compress_tuple_threshold = Min(compress_tuple_threshold,
+								   toast_tuple_threshold);
+
 	/*
 	 * If the new tuple is too big for storage or contains already toasted
 	 * out-of-line attributes from some other relation, invoke the toaster.
+	 *
+	 * The toaster is invoked only if the tuple length is greater than the
+	 * compression limit.  Note that compress_tuple_threshold must be less
+	 * than or equal to toast_tuple_threshold, so it's enough to only test
+	 * for compress_tuple_threshold.
 	 */
 	if (relation->rd_rel->relkind != RELKIND_RELATION &&
 		relation->rd_rel->relkind != RELKIND_MATVIEW)
@@ -2088,7 +2107,8 @@ heap_prepare_insert(Relation relation, HeapTuple tup, TransactionId xid,
 		Assert(!HeapTupleHasExternal(tup));
 		return tup;
 	}
-	else if (HeapTupleHasExternal(tup) || tup->t_len > TOAST_TUPLE_THRESHOLD)
+	else if (HeapTupleHasExternal(tup) ||
+			 tup->t_len > compress_tuple_threshold)
 		return toast_insert_or_update(relation, tup, NULL, options);
 	else
 		return tup;
@@ -3390,9 +3410,30 @@ l2:
 		need_toast = false;
 	}
 	else
+	{
+		int			toast_tuple_threshold;
+		int			compress_tuple_threshold;
+
+		/*
+		 * Compute the compressibility threshold, which is the minimum between
+		 * toast_tuple_target and compress_tuple_target for a relation.
+		 */
+		toast_tuple_threshold = RelationGetToastTupleTarget(relation,
+									TOAST_TUPLE_TARGET);
+		compress_tuple_threshold = RelationGetCompressTupleTarget(relation,
+									COMPRESS_TUPLE_TARGET);
+		compress_tuple_threshold = Min(compress_tuple_threshold,
+									   toast_tuple_threshold);
+
+		/*
+		 * compress_tuple_threshold must be less than or equal to
+		 * toast_tuple_threshold, so it is enough to only test
+		 * compress_tuple_threshold here.
+		 */
 		need_toast = (HeapTupleHasExternal(&oldtup) ||
 					  HeapTupleHasExternal(newtup) ||
-					  newtup->t_len > TOAST_TUPLE_THRESHOLD);
+					  newtup->t_len > compress_tuple_threshold);
+	}
 
 	pagefree = PageGetHeapFreeSpace(page);
 
diff --git a/src/backend/access/heap/rewriteheap.c b/src/backend/access/heap/rewriteheap.c
index bce4274362..a7d5a29d36 100644
--- a/src/backend/access/heap/rewriteheap.c
+++ b/src/backend/access/heap/rewriteheap.c
@@ -636,11 +636,29 @@ raw_heap_insert(RewriteState state, HeapTuple tup)
 	Size		len;
 	OffsetNumber newoff;
 	HeapTuple	heaptup;
+	int			toast_tuple_threshold;
+	int			compress_tuple_threshold;
+
+	/*
+	 * Compute the compressibility threshold, which is the minimum between
+	 * toast_tuple_target and compress_tuple_target for a relation.
+	 */
+	toast_tuple_threshold = RelationGetToastTupleTarget(state->rs_new_rel,
+									TOAST_TUPLE_TARGET);
+	compress_tuple_threshold = RelationGetCompressTupleTarget(state->rs_new_rel,
+									COMPRESS_TUPLE_TARGET);
+	compress_tuple_threshold = Min(compress_tuple_threshold,
+								   toast_tuple_threshold);
 
 	/*
 	 * If the new tuple is too big for storage or contains already toasted
 	 * out-of-line attributes from some other relation, invoke the toaster.
 	 *
+	 * The toaster is invoked only if the tuple length is greater than the
+	 * compression limit.  Note that compress_tuple_threshold must be less
+	 * than or equal to toast_tuple_threshold, so it's enough to only test
+	 * for compress_tuple_threshold.
+	 *
 	 * Note: below this point, heaptup is the data we actually intend to store
 	 * into the relation; tup is the caller's original untoasted data.
 	 */
@@ -650,7 +668,8 @@ raw_heap_insert(RewriteState state, HeapTuple tup)
 		Assert(!HeapTupleHasExternal(tup));
 		heaptup = tup;
 	}
-	else if (HeapTupleHasExternal(tup) || tup->t_len > TOAST_TUPLE_THRESHOLD)
+	else if (HeapTupleHasExternal(tup) ||
+			 tup->t_len > compress_tuple_threshold)
 	{
 		int options = HEAP_INSERT_SKIP_FSM;
 
diff --git a/src/backend/access/heap/tuptoaster.c b/src/backend/access/heap/tuptoaster.c
index 74e957abb7..2bfbc4195a 100644
--- a/src/backend/access/heap/tuptoaster.c
+++ b/src/backend/access/heap/tuptoaster.c
@@ -549,6 +549,7 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
 	bool		has_nulls = false;
 
 	Size		maxDataLen;
+	Size		maxCompressLen;
 	Size		hoff;
 
 	char		toast_action[MaxHeapAttributeNumber];
@@ -731,12 +732,19 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
 	/* now convert to a limit on the tuple data size */
 	maxDataLen = RelationGetToastTupleTarget(rel, TOAST_TUPLE_TARGET) - hoff;
 
+	/*
+	 * Get the limit at which we should apply compression. This will be same as
+	 * maxDataLen unless overridden by the user explicitly.
+	 */
+	maxCompressLen = RelationGetCompressTupleTarget(rel, COMPRESS_TUPLE_TARGET) - hoff;
+	maxCompressLen = Min(maxCompressLen, maxDataLen);
+
 	/*
 	 * Look for attributes with attstorage 'x' to compress.  Also find large
 	 * attributes with attstorage 'x' or 'e', and store them external.
 	 */
 	while (heap_compute_data_size(tupleDesc,
-								  toast_values, toast_isnull) > maxDataLen)
+								  toast_values, toast_isnull) > maxCompressLen)
 	{
 		int			biggest_attno = -1;
 		int32		biggest_size = MAXALIGN(TOAST_POINTER_SIZE);
@@ -881,7 +889,7 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
 	 * compression
 	 */
 	while (heap_compute_data_size(tupleDesc,
-								  toast_values, toast_isnull) > maxDataLen)
+								  toast_values, toast_isnull) > maxCompressLen)
 	{
 		int			biggest_attno = -1;
 		int32		biggest_size = MAXALIGN(TOAST_POINTER_SIZE);
diff --git a/src/include/access/tuptoaster.h b/src/include/access/tuptoaster.h
index 4bfefffbf3..0d6e7c43da 100644
--- a/src/include/access/tuptoaster.h
+++ b/src/include/access/tuptoaster.h
@@ -33,10 +33,19 @@
 				  / (tuplesPerPage))
 
 /*
- * These symbols control toaster activation.  If a tuple is larger than
- * TOAST_TUPLE_THRESHOLD, we will try to toast it down to no more than
- * TOAST_TUPLE_TARGET bytes through compressing compressible fields and
- * moving EXTENDED and EXTERNAL data out-of-line.
+ * These symbols control toaster and compression activation.  If a tuple
+ * is larger than COMPRESS_TUPLE_TARGET, we will try to compress it first
+ * if its column is marked as MAIN or EXTENDED.  If the compressed row is
+ * still greater than TOAST_TUPLE_TARGET, then we will try to toast it
+ * down to TOAST_TUPLE_TARGET bytes through compressing compressible
+ * fields and moving EXTENDED and EXTERNAL data out-of-line.
+ *
+ * By default, COMPRESS_TUPLE_TARGET and TOAST_TUPLE_TARGET have the same
+ * threshold values, meaning that compression will only be tried when toasting
+ * a tuple.  It is possible to tune that behavior using the relation option
+ * COMPRESS_TUPLE_TARGET.  If COMPRESS_TUPLE_TARGET is higher than
+ * TOAST_TUPLE_TARGET, then only TOAST_TUPLE_TARGET is used to decide if a
+ * tuple is toasted and compressed.
  *
  * The numbers need not be the same, though they currently are.  It doesn't
  * make sense for TARGET to exceed THRESHOLD, but it could be useful to make
@@ -56,6 +65,8 @@
 
 #define TOAST_TUPLE_TARGET		TOAST_TUPLE_THRESHOLD
 
+#define COMPRESS_TUPLE_TARGET	TOAST_TUPLE_THRESHOLD
+
 /*
  * The code will also consider moving MAIN data out-of-line, but only as a
  * last resort if the previous steps haven't reached the target tuple size.
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index 54028515a7..9e17e2a599 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -263,6 +263,7 @@ typedef struct StdRdOptions
 	/* fraction of newly inserted tuples prior to trigger index cleanup */
 	float8		vacuum_cleanup_index_scale_factor;
 	int			toast_tuple_target; /* target for tuple toasting */
+	int			compress_tuple_target; /* target for tuple compression */
 	AutoVacOpts autovacuum;		/* autovacuum-related options */
 	bool		user_catalog_table; /* use as an additional catalog relation */
 	int			parallel_workers;	/* max number of parallel workers */
@@ -279,6 +280,14 @@ typedef struct StdRdOptions
 	((relation)->rd_options ? \
 	 ((StdRdOptions *) (relation)->rd_options)->toast_tuple_target : (defaulttarg))
 
+/*
+ * RelationGetCompressTupleTarget
+ *		Returns the relation's compress_tuple_target.  Note multiple eval of argument!
+ */
+#define RelationGetCompressTupleTarget(relation, defaulttarg) \
+	((relation)->rd_options ? \
+	 ((StdRdOptions *) (relation)->rd_options)->compress_tuple_target : (defaulttarg))
+
 /*
  * RelationGetFillFactor
  *		Returns the relation's fillfactor.  Note multiple eval of argument!
diff --git a/src/test/regress/expected/strings.out b/src/test/regress/expected/strings.out
index 189bdffdca..66757ba8c4 100644
--- a/src/test/regress/expected/strings.out
+++ b/src/test/regress/expected/strings.out
@@ -1209,7 +1209,69 @@ select 0 = pg_relation_size('pg_toast.pg_toast_'||(select oid from pg_class wher
  t
 (1 row)
 
+-- Test tuple compression with compress_tuple_target
+CREATE TABLE compresstest250 (a int, b text)
+  WITH (compress_tuple_target = 250);
+CREATE TABLE compresstest2040 (a int, b text)
+  WITH (compress_tuple_target = 2040);
+-- tuple should not get compressed nor toasted
+INSERT INTO compresstest250 VALUES (1, repeat('1234567890',20));
+-- tuple should get compressed but not toasted
+INSERT INTO compresstest250 VALUES (2, repeat('1234567890',30));
+-- tuple should not get compressed nor toasted
+INSERT INTO compresstest2040 VALUES (1, repeat('1234567890',20));
+-- tuple should not get compressed nor toasted
+INSERT INTO compresstest2040 VALUES (2, repeat('1234567890',30));
+SELECT a, pg_column_size(b) < 250 FROM compresstest250 ORDER BY a;
+ a | ?column? 
+---+----------
+ 1 | t
+ 2 | t
+(2 rows)
+
+SELECT a, pg_column_size(b) < 250 FROM compresstest2040 ORDER BY a;
+ a | ?column? 
+---+----------
+ 1 | t
+ 2 | f
+(2 rows)
+
+-- expect 0 blocks in toast relation
+SELECT pg_relation_size(reltoastrelid) = 0 AS data_size
+  FROM pg_class WHERE relname = 'compresstest250';
+ data_size 
+-----------
+ t
+(1 row)
+
+SELECT pg_relation_size(reltoastrelid) = 0 AS data_size
+  FROM pg_class WHERE relname = 'compresstest2040';
+ data_size 
+-----------
+ t
+(1 row)
+
+-- tuple should get compressed and toasted
+INSERT INTO compresstest250 SELECT 3, string_agg('', md5(random()::text))
+  FROM generate_series(1,300);
+SELECT a, pg_column_size(b) < 250 FROM compresstest250 ORDER by a;
+ a | ?column? 
+---+----------
+ 1 | t
+ 2 | t
+ 3 | f
+(3 rows)
+
+-- expect > 0 blocks in toast relation
+SELECT pg_relation_size(reltoastrelid) = 0 AS data_size
+  FROM pg_class WHERE relname = 'compresstest250';
+ data_size 
+-----------
+ f
+(1 row)
+
 DROP TABLE toasttest;
+DROP TABLE compresstest250, compresstest2040;
 --
 -- test substr with toasted bytea values
 --
diff --git a/src/test/regress/sql/strings.sql b/src/test/regress/sql/strings.sql
index f2203ef1b1..e0ce45dd68 100644
--- a/src/test/regress/sql/strings.sql
+++ b/src/test/regress/sql/strings.sql
@@ -386,7 +386,36 @@ INSERT INTO toasttest values (repeat('1234567890',300));
 -- expect 0 blocks
 select 0 = pg_relation_size('pg_toast.pg_toast_'||(select oid from pg_class where relname = 'toasttest'))/current_setting('block_size')::integer as blocks;
 
+-- Test tuple compression with compress_tuple_target
+CREATE TABLE compresstest250 (a int, b text)
+  WITH (compress_tuple_target = 250);
+CREATE TABLE compresstest2040 (a int, b text)
+  WITH (compress_tuple_target = 2040);
+-- tuple should not get compressed nor toasted
+INSERT INTO compresstest250 VALUES (1, repeat('1234567890',20));
+-- tuple should get compressed but not toasted
+INSERT INTO compresstest250 VALUES (2, repeat('1234567890',30));
+-- tuple should not get compressed nor toasted
+INSERT INTO compresstest2040 VALUES (1, repeat('1234567890',20));
+-- tuple should not get compressed nor toasted
+INSERT INTO compresstest2040 VALUES (2, repeat('1234567890',30));
+SELECT a, pg_column_size(b) < 250 FROM compresstest250 ORDER BY a;
+SELECT a, pg_column_size(b) < 250 FROM compresstest2040 ORDER BY a;
+-- expect 0 blocks in toast relation
+SELECT pg_relation_size(reltoastrelid) = 0 AS data_size
+  FROM pg_class WHERE relname = 'compresstest250';
+SELECT pg_relation_size(reltoastrelid) = 0 AS data_size
+  FROM pg_class WHERE relname = 'compresstest2040';
+-- tuple should get compressed and toasted
+INSERT INTO compresstest250 SELECT 3, string_agg('', md5(random()::text))
+  FROM generate_series(1,300);
+SELECT a, pg_column_size(b) < 250 FROM compresstest250 ORDER by a;
+-- expect > 0 blocks in toast relation
+SELECT pg_relation_size(reltoastrelid) = 0 AS data_size
+  FROM pg_class WHERE relname = 'compresstest250';
+
 DROP TABLE toasttest;
+DROP TABLE compresstest250, compresstest2040;
 
 --
 -- test substr with toasted bytea values