Replicate generated columns when 'publish_generated_columns' is set.

This patch builds on the work done in commit 745217a051 by enabling the
replication of generated columns alongside regular column changes through
a new publication parameter: publish_generated_columns.

Example usage:
CREATE PUBLICATION pub1 FOR TABLE tab_gencol WITH (publish_generated_columns = true);

The column list takes precedence. If the generated columns are specified
in the column list, they will be replicated even if
'publish_generated_columns' is set to false. Conversely, if generated
columns are not included in the column list (assuming the user specifies a
column list), they will not be replicated even if
'publish_generated_columns' is true.

Author: Vignesh C, Shubham Khanna
Reviewed-by: Peter Smith, Amit Kapila, Hayato Kuroda, Shlok Kyal, Ajin Cherian, Hou Zhijie, Masahiko Sawada
Discussion: https://postgr.es/m/B80D17B2-2C8E-4C7D-87F2-E5B4BE3C069E@gmail.com

1 files changed, 35 insertions, 34 deletions

diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index ac4af53feba..2c2085b2f98 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -30,10 +30,11 @@
 #define TRUNCATE_RESTART_SEQS	(1<<1)
 
 static void logicalrep_write_attrs(StringInfo out, Relation rel,
-								   Bitmapset *columns);
+								   Bitmapset *columns, bool include_gencols);
 static void logicalrep_write_tuple(StringInfo out, Relation rel,
 								   TupleTableSlot *slot,
-								   bool binary, Bitmapset *columns);
+								   bool binary, Bitmapset *columns,
+								   bool include_gencols);
 static void logicalrep_read_attrs(StringInfo in, LogicalRepRelation *rel);
 static void logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple);
 
@@ -399,7 +400,8 @@ logicalrep_read_origin(StringInfo in, XLogRecPtr *origin_lsn)
  */
 void
 logicalrep_write_insert(StringInfo out, TransactionId xid, Relation rel,
-						TupleTableSlot *newslot, bool binary, Bitmapset *columns)
+						TupleTableSlot *newslot, bool binary,
+						Bitmapset *columns, bool include_gencols)
 {
 	pq_sendbyte(out, LOGICAL_REP_MSG_INSERT);
 
@@ -411,7 +413,7 @@ logicalrep_write_insert(StringInfo out, TransactionId xid, Relation rel,
 	pq_sendint32(out, RelationGetRelid(rel));
 
 	pq_sendbyte(out, 'N');		/* new tuple follows */
-	logicalrep_write_tuple(out, rel, newslot, binary, columns);
+	logicalrep_write_tuple(out, rel, newslot, binary, columns, include_gencols);
 }
 
 /*
@@ -444,7 +446,7 @@ logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup)
 void
 logicalrep_write_update(StringInfo out, TransactionId xid, Relation rel,
 						TupleTableSlot *oldslot, TupleTableSlot *newslot,
-						bool binary, Bitmapset *columns)
+						bool binary, Bitmapset *columns, bool include_gencols)
 {
 	pq_sendbyte(out, LOGICAL_REP_MSG_UPDATE);
 
@@ -465,11 +467,12 @@ logicalrep_write_update(StringInfo out, TransactionId xid, Relation rel,
 			pq_sendbyte(out, 'O');	/* old tuple follows */
 		else
 			pq_sendbyte(out, 'K');	/* old key follows */
-		logicalrep_write_tuple(out, rel, oldslot, binary, columns);
+		logicalrep_write_tuple(out, rel, oldslot, binary, columns,
+							   include_gencols);
 	}
 
 	pq_sendbyte(out, 'N');		/* new tuple follows */
-	logicalrep_write_tuple(out, rel, newslot, binary, columns);
+	logicalrep_write_tuple(out, rel, newslot, binary, columns, include_gencols);
 }
 
 /*
@@ -519,7 +522,7 @@ logicalrep_read_update(StringInfo in, bool *has_oldtuple,
 void
 logicalrep_write_delete(StringInfo out, TransactionId xid, Relation rel,
 						TupleTableSlot *oldslot, bool binary,
-						Bitmapset *columns)
+						Bitmapset *columns, bool include_gencols)
 {
 	Assert(rel->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT ||
 		   rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL ||
@@ -539,7 +542,7 @@ logicalrep_write_delete(StringInfo out, TransactionId xid, Relation rel,
 	else
 		pq_sendbyte(out, 'K');	/* old key follows */
 
-	logicalrep_write_tuple(out, rel, oldslot, binary, columns);
+	logicalrep_write_tuple(out, rel, oldslot, binary, columns, include_gencols);
 }
 
 /*
@@ -655,7 +658,7 @@ logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn,
  */
 void
 logicalrep_write_rel(StringInfo out, TransactionId xid, Relation rel,
-					 Bitmapset *columns)
+					 Bitmapset *columns, bool include_gencols)
 {
 	char	   *relname;
 
@@ -677,7 +680,7 @@ logicalrep_write_rel(StringInfo out, TransactionId xid, Relation rel,
 	pq_sendbyte(out, rel->rd_rel->relreplident);
 
 	/* send the attribute info */
-	logicalrep_write_attrs(out, rel, columns);
+	logicalrep_write_attrs(out, rel, columns, include_gencols);
 }
 
 /*
@@ -754,7 +757,7 @@ logicalrep_read_typ(StringInfo in, LogicalRepTyp *ltyp)
  */
 static void
 logicalrep_write_tuple(StringInfo out, Relation rel, TupleTableSlot *slot,
-					   bool binary, Bitmapset *columns)
+					   bool binary, Bitmapset *columns, bool include_gencols)
 {
 	TupleDesc	desc;
 	Datum	   *values;
@@ -768,7 +771,7 @@ logicalrep_write_tuple(StringInfo out, Relation rel, TupleTableSlot *slot,
 	{
 		Form_pg_attribute att = TupleDescAttr(desc, i);
 
-		if (!logicalrep_should_publish_column(att, columns))
+		if (!logicalrep_should_publish_column(att, columns, include_gencols))
 			continue;
 
 		nliveatts++;
@@ -786,7 +789,7 @@ logicalrep_write_tuple(StringInfo out, Relation rel, TupleTableSlot *slot,
 		Form_pg_type typclass;
 		Form_pg_attribute att = TupleDescAttr(desc, i);
 
-		if (!logicalrep_should_publish_column(att, columns))
+		if (!logicalrep_should_publish_column(att, columns, include_gencols))
 			continue;
 
 		if (isnull[i])
@@ -904,7 +907,8 @@ logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple)
  * Write relation attribute metadata to the stream.
  */
 static void
-logicalrep_write_attrs(StringInfo out, Relation rel, Bitmapset *columns)
+logicalrep_write_attrs(StringInfo out, Relation rel, Bitmapset *columns,
+					   bool include_gencols)
 {
 	TupleDesc	desc;
 	int			i;
@@ -919,7 +923,7 @@ logicalrep_write_attrs(StringInfo out, Relation rel, Bitmapset *columns)
 	{
 		Form_pg_attribute att = TupleDescAttr(desc, i);
 
-		if (!logicalrep_should_publish_column(att, columns))
+		if (!logicalrep_should_publish_column(att, columns, include_gencols))
 			continue;
 
 		nliveatts++;
@@ -937,7 +941,7 @@ logicalrep_write_attrs(StringInfo out, Relation rel, Bitmapset *columns)
 		Form_pg_attribute att = TupleDescAttr(desc, i);
 		uint8		flags = 0;
 
-		if (!logicalrep_should_publish_column(att, columns))
+		if (!logicalrep_should_publish_column(att, columns, include_gencols))
 			continue;
 
 		/* REPLICA IDENTITY FULL means all columns are sent as part of key. */
@@ -1248,29 +1252,26 @@ logicalrep_message_type(LogicalRepMsgType action)
 /*
  * Check if the column 'att' of a table should be published.
  *
- * 'columns' represents the column list specified for that table in the
- * publication.
+ * 'columns' represents the publication column list (if any) for that table.
  *
- * Note that generated columns can be present only in 'columns' list.
+ * 'include_gencols' flag indicates whether generated columns should be
+ * published when there is no column list. Typically, this will have the same
+ * value as the 'publish_generated_columns' publication parameter.
+ *
+ * Note that generated columns can be published only when present in a
+ * publication column list, or when include_gencols is true.
  */
 bool
-logicalrep_should_publish_column(Form_pg_attribute att, Bitmapset *columns)
+logicalrep_should_publish_column(Form_pg_attribute att, Bitmapset *columns,
+								 bool include_gencols)
 {
 	if (att->attisdropped)
 		return false;
 
-	/*
-	 * Skip publishing generated columns if they are not included in the
-	 * column list.
-	 */
-	if (!columns && att->attgenerated)
-		return false;
-
-	/*
-	 * Check if a column is covered by a column list.
-	 */
-	if (columns && !bms_is_member(att->attnum, columns))
-		return false;
+	/* If a column list is provided, publish only the cols in that list. */
+	if (columns)
+		return bms_is_member(att->attnum, columns);
 
-	return true;
+	/* All non-generated columns are always published. */
+	return att->attgenerated ? include_gencols : true;
 }