Allow specifying column lists for logical replication

This allows specifying an optional column list when adding a table to
logical replication. The column list may be specified after the table
name, enclosed in parentheses. Columns not included in this list are not
sent to the subscriber, allowing the schema on the subscriber to be a
subset of the publisher schema.

For UPDATE/DELETE publications, the column list needs to cover all
REPLICA IDENTITY columns. For INSERT publications, the column list is
arbitrary and may omit some REPLICA IDENTITY columns. Furthermore, if
the table uses REPLICA IDENTITY FULL, column list is not allowed.

The column list can contain only simple column references. Complex
expressions, function calls etc. are not allowed. This restriction could
be relaxed in the future.

During the initial table synchronization, only columns included in the
column list are copied to the subscriber. If the subscription has
several publications, containing the same table with different column
lists, columns specified in any of the lists will be copied.

This means all columns are replicated if the table has no column list
at all (which is treated as column list with all columns), or when of
the publications is defined as FOR ALL TABLES (possibly IN SCHEMA that
matches the schema of the table).

For partitioned tables, publish_via_partition_root determines whether
the column list for the root or the leaf relation will be used. If the
parameter is 'false' (the default), the list defined for the leaf
relation is used. Otherwise, the column list for the root partition
will be used.

Psql commands \dRp+ and \d <table-name> now display any column lists.

Author: Tomas Vondra, Alvaro Herrera, Rahila Syed
Reviewed-by: Peter Eisentraut, Alvaro Herrera, Vignesh C, Ibrar Ahmed,
Amit Kapila, Hou zj, Peter Smith, Wang wei, Tang, Shi yu
Discussion: https://postgr.es/m/CAH2L28vddB_NFdRVpuyRBJEBWjz4BSyTB=_ektNRH8NJ1jf95g@mail.gmail.com

5 files changed, 20 insertions, 4 deletions

diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h
index 97f26208e1d..186d8ea74b6 100644
--- a/src/include/catalog/pg_publication.h
+++ b/src/include/catalog/pg_publication.h
@@ -95,6 +95,13 @@ typedef struct PublicationDesc
 	 */
 	bool		rf_valid_for_update;
 	bool		rf_valid_for_delete;
+
+	/*
+	 * true if the columns are part of the replica identity or the publication actions
+	 * do not include UPDATE or DELETE.
+	 */
+	bool		cols_valid_for_update;
+	bool		cols_valid_for_delete;
 } PublicationDesc;
 
 typedef struct Publication
@@ -111,6 +118,7 @@ typedef struct PublicationRelInfo
 {
 	Relation	relation;
 	Node	   *whereClause;
+	List	   *columns;
 } PublicationRelInfo;
 
 extern Publication *GetPublication(Oid pubid);
@@ -137,6 +145,7 @@ extern List *GetPublicationRelations(Oid pubid, char objectType,
 									 PublicationPartOpt pub_partopt);
 extern List *GetAllTablesPublications(void);
 extern List *GetAllTablesPublicationRelations(bool pubviaroot);
+extern void GetActionsInPublication(Oid pubid, PublicationActions *actions);
 extern List *GetPublicationSchemas(Oid pubid, char objectType);
 extern List *GetSchemaPublications(Oid schemaid, char objectType);
 extern List *GetSchemaPublicationRelations(Oid schemaid, char objectType,
@@ -160,6 +169,9 @@ extern ObjectAddress publication_add_schema(Oid pubid, Oid schemaid,
 											char objectType,
 											bool if_not_exists);
 
+extern Bitmapset *pub_collist_to_bitmapset(Bitmapset *columns, Datum pubcols,
+										   MemoryContext mcxt);
+
 extern Oid	get_publication_oid(const char *pubname, bool missing_ok);
 extern char *get_publication_name(Oid pubid, bool missing_ok);
 
diff --git a/src/include/catalog/pg_publication_rel.h b/src/include/catalog/pg_publication_rel.h
index 0dd0f425db9..4feb581899e 100644
--- a/src/include/catalog/pg_publication_rel.h
+++ b/src/include/catalog/pg_publication_rel.h
@@ -34,6 +34,7 @@ CATALOG(pg_publication_rel,6106,PublicationRelRelationId)
 
 #ifdef	CATALOG_VARLEN			/* variable-length fields start here */
 	pg_node_tree prqual;		/* qualifications */
+	int2vector	prattrs;		/* columns to replicate */
 #endif
 } FormData_pg_publication_rel;
 
diff --git a/src/include/commands/publicationcmds.h b/src/include/commands/publicationcmds.h
index 7813cbcb6bb..ae87caf089d 100644
--- a/src/include/commands/publicationcmds.h
+++ b/src/include/commands/publicationcmds.h
@@ -31,7 +31,9 @@ extern void RemovePublicationSchemaById(Oid psoid);
 extern ObjectAddress AlterPublicationOwner(const char *name, Oid newOwnerId);
 extern void AlterPublicationOwner_oid(Oid pubid, Oid newOwnerId);
 extern void InvalidatePublicationRels(List *relids);
-extern bool contain_invalid_rfcolumn(Oid pubid, Relation relation,
+extern bool pub_rf_contains_invalid_column(Oid pubid, Relation relation,
+									 List *ancestors, bool pubviaroot);
+extern bool pub_collist_contains_invalid_column(Oid pubid, Relation relation,
 									 List *ancestors, bool pubviaroot);
 
 #endif							/* PUBLICATIONCMDS_H */
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index cb1fcc0ee31..5a458c42e5e 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -3654,6 +3654,7 @@ typedef struct PublicationTable
 	NodeTag		type;
 	RangeVar   *relation;		/* relation to be published */
 	Node	   *whereClause;	/* qualifications */
+	List	   *columns;		/* List of columns in a publication table */
 } PublicationTable;
 
 /*
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index fb86ca022d2..13ee10fdd4e 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -222,12 +222,12 @@ extern char *logicalrep_read_origin(StringInfo in, XLogRecPtr *origin_lsn);
 extern void logicalrep_write_insert(StringInfo out, TransactionId xid,
 									Relation rel,
 									TupleTableSlot *newslot,
-									bool binary);
+									bool binary, Bitmapset *columns);
 extern LogicalRepRelId logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup);
 extern void logicalrep_write_update(StringInfo out, TransactionId xid,
 									Relation rel,
 									TupleTableSlot *oldslot,
-									TupleTableSlot *newslot, bool binary);
+									TupleTableSlot *newslot, bool binary, Bitmapset *columns);
 extern LogicalRepRelId logicalrep_read_update(StringInfo in,
 											  bool *has_oldtuple, LogicalRepTupleData *oldtup,
 											  LogicalRepTupleData *newtup);
@@ -250,7 +250,7 @@ extern void logicalrep_write_sequence(StringInfo out, Relation rel,
 									  bool is_called);
 extern void logicalrep_read_sequence(StringInfo in, LogicalRepSequence *seqdata);
 extern void logicalrep_write_rel(StringInfo out, TransactionId xid,
-								 Relation rel);
+								 Relation rel, Bitmapset *columns);
 extern LogicalRepRelation *logicalrep_read_rel(StringInfo in);
 extern void logicalrep_write_typ(StringInfo out, TransactionId xid,
 								 Oid typoid);