column-filtering-8.patch

diff --git a/src/backend/catalog/dependency.c b/src/backend/catalog/dependency.c
index fe9c714257..a88d12e8ae 100644
--- a/src/backend/catalog/dependency.c
+++ b/src/backend/catalog/dependency.c
@@ -1472,7 +1472,7 @@ doDeletion(const ObjectAddress *object, int flags)
 			break;
 
 		case OCLASS_PUBLICATION_REL:
-			RemovePublicationRelById(object->objectId);
+			RemovePublicationRelById(object->objectId, object->objectSubId);
 			break;
 
 		case OCLASS_PUBLICATION:
@@ -2754,8 +2754,12 @@ free_object_addresses(ObjectAddresses *addrs)
 ObjectClass
 getObjectClass(const ObjectAddress *object)
 {
-	/* only pg_class entries can have nonzero objectSubId */
+	/*
+	 * only pg_class and pg_publication_rel entries can have nonzero
+	 * objectSubId
+	 */
 	if (object->classId != RelationRelationId &&
+		object->classId != PublicationRelRelationId &&
 		object->objectSubId != 0)
 		elog(ERROR, "invalid non-zero objectSubId for object class %u",
 			 object->classId);
diff --git a/src/backend/catalog/objectaddress.c b/src/backend/catalog/objectaddress.c
index 2bae3fbb17..5eed248dcb 100644
--- a/src/backend/catalog/objectaddress.c
+++ b/src/backend/catalog/objectaddress.c
@@ -4019,6 +4019,7 @@ getObjectDescription(const ObjectAddress *object, bool missing_ok)
 				/* translator: first %s is, e.g., "table %s" */
 				appendStringInfo(&buffer, _("publication of %s in publication %s"),
 								 rel.data, pubname);
+				/* FIXME add objectSubId support */
 				pfree(rel.data);
 				ReleaseSysCache(tup);
 				break;
@@ -5853,9 +5854,16 @@ getObjectIdentityParts(const ObjectAddress *object,
 
 				getRelationIdentity(&buffer, prform->prrelid, objname, false);
 				appendStringInfo(&buffer, " in publication %s", pubname);
+				if (object->objectSubId)	/* FIXME maybe get_attname */
+					appendStringInfo(&buffer, " column %d", object->objectSubId);
 
 				if (objargs)
+				{
 					*objargs = list_make1(pubname);
+					if (object->objectSubId)
+						*objargs = lappend(*objargs,
+										   psprintf("%d", object->objectSubId));
+				}
 
 				ReleaseSysCache(tup);
 				break;
diff --git a/src/backend/catalog/pg_depend.c b/src/backend/catalog/pg_depend.c
index 07bcdc463a..462c8efe70 100644
--- a/src/backend/catalog/pg_depend.c
+++ b/src/backend/catalog/pg_depend.c
@@ -658,6 +658,56 @@ isObjectPinned(const ObjectAddress *object)
  * Various special-purpose lookups and manipulations of pg_depend.
  */
 
+/*
+ * Find all objects of the given class that reference the specified object,
+ * and add them to the given ObjectAddresses.
+ */
+void
+findAndAddAddresses(ObjectAddresses *addrs, Oid classId,
+					Oid refclassId, Oid refobjectId, int32 refobjsubId)
+{
+	Relation	depRel;
+	ScanKeyData	key[3];
+	SysScanDesc scan;
+	HeapTuple	tup;
+
+	depRel = table_open(DependRelationId, AccessShareLock);
+
+	ScanKeyInit(&key[0],
+				Anum_pg_depend_refclassid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(refclassId));
+	ScanKeyInit(&key[1],
+				Anum_pg_depend_refobjid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(refobjectId));
+	ScanKeyInit(&key[2],
+				Anum_pg_depend_refobjsubid,
+				BTEqualStrategyNumber, F_INT4EQ,
+				Int32GetDatum(refobjsubId));
+
+	scan = systable_beginscan(depRel, DependReferenceIndexId, true,
+							  NULL, 3, key);
+
+	while (HeapTupleIsValid(tup = systable_getnext(scan)))
+	{
+		Form_pg_depend depform = (Form_pg_depend) GETSTRUCT(tup);
+		ObjectAddress	object;
+
+		if (depform->classid != classId)
+			continue;
+
+		ObjectAddressSubSet(object, depform->classid, depform->objid,
+							depform->refobjsubid);
+
+		add_exact_object_address(&object, addrs);
+	}
+
+	systable_endscan(scan);
+
+	table_close(depRel, AccessShareLock);
+}
+
 
 /*
  * Find the extension containing the specified object, if any
diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index a5229aea51..b99b3748c9 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -328,6 +328,8 @@ publication_add_relation(Oid pubid, PublicationRelInfo *targetrel,
 	{
 		table_close(rel, RowExclusiveLock);
 
+		/* FIXME need to handle the case of different column list */
+
 		if (if_not_exists)
 			return InvalidObjectAddress;
 
@@ -395,12 +397,6 @@ publication_add_relation(Oid pubid, PublicationRelInfo *targetrel,
 
 	ObjectAddressSet(myself, PublicationRelRelationId, prrelid);
 
-	while ((attnum = bms_first_member(attmap)) >= 0)
-	{
-		/* Add dependency on the column */
-		ObjectAddressSubSet(referenced, RelationRelationId, relid, attnum);
-		recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
-	}
 	/* Add dependency on the publication */
 	ObjectAddressSet(referenced, PublicationRelationId, pubid);
 	recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
@@ -409,6 +405,21 @@ publication_add_relation(Oid pubid, PublicationRelInfo *targetrel,
 	ObjectAddressSet(referenced, RelationRelationId, relid);
 	recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
 
+	/*
+	 * If there's an explicit column list, make one dependency entry for each
+	 * column.  Note that the referencing side of the dependency is also
+	 * specific to one column, so that it can be dropped separately if the
+	 * column is dropped.
+	 */
+	while ((attnum = bms_first_member(attmap)) >= 0)
+	{
+		ObjectAddressSubSet(referenced, RelationRelationId, relid,
+							attnum + FirstLowInvalidHeapAttributeNumber);
+		myself.objectSubId = attnum + FirstLowInvalidHeapAttributeNumber;
+		recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
+	}
+	myself.objectSubId = 0;		/* need to undo this bit */
+
 	/* Close the table. */
 	table_close(rel, RowExclusiveLock);
 
diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
index 753df44613..0078c5986c 100644
--- a/src/backend/commands/publicationcmds.c
+++ b/src/backend/commands/publicationcmds.c
@@ -561,7 +561,6 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup,
 
 				pubrel = palloc(sizeof(PublicationRelInfo));
 				pubrel->relation = oldrel;
-				/* This is not needed to delete a table */
 				pubrel->columns = NIL;
 				delrels = lappend(delrels, pubrel);
 			}
@@ -758,10 +757,11 @@ AlterPublication(ParseState *pstate, AlterPublicationStmt *stmt)
 }
 
 /*
- * Remove relation from publication by mapping OID.
+ * Remove relation from publication by mapping OID, or publication status
+ * of one column of that relation in the publication if an attnum is given.
  */
 void
-RemovePublicationRelById(Oid proid)
+RemovePublicationRelById(Oid proid, int32 attnum)
 {
 	Relation	rel;
 	HeapTuple	tup;
@@ -791,7 +791,81 @@ RemovePublicationRelById(Oid proid)
 
 	InvalidatePublicationRels(relids);
 
-	CatalogTupleDelete(rel, &tup->t_self);
+	/*
+	 * If no column is given, simply delete the relation from the publication.
+	 *
+	 * If a column is given, what we do instead is to remove that column from
+	 * the column list.  The relation remains in the publication, with the
+	 * other columns.  However, dropping the last column is disallowed.
+	 */
+	if (attnum == 0)
+	{
+		CatalogTupleDelete(rel, &tup->t_self);
+	}
+	else
+	{
+		Datum		adatum;
+		ArrayType  *arr;
+		int			nelems;
+		int16	   *elems;
+		int16	   *newelems;
+		int2vector *newvec;
+		Datum		values[Natts_pg_publication_rel];
+		bool		nulls[Natts_pg_publication_rel];
+		bool		replace[Natts_pg_publication_rel];
+		HeapTuple	newtup;
+		int			i,
+					j;
+		bool		isnull;
+
+		/* Obtain the original column list */
+		adatum = SysCacheGetAttr(PUBLICATIONRELMAP,
+								 tup,
+								 Anum_pg_publication_rel_prattrs,
+								 &isnull);
+		if (isnull)			/* shouldn't happen */
+			elog(ERROR, "can't drop column from publication without a column list");
+		arr = DatumGetArrayTypeP(adatum);
+		nelems = ARR_DIMS(arr)[0];
+		elems = (int16 *) ARR_DATA_PTR(arr);
+
+		/* Construct a list excluding the given column */
+		newelems = palloc(sizeof(int16) * nelems - 1);
+		for (i = 0, j = 0; i < nelems - 1; i++)
+		{
+			if (elems[i] == attnum)
+				continue;
+			newelems[j++] = elems[i];
+		}
+
+		/*
+		 * If this is the last column used in the publication, disallow the
+		 * command. We could alternatively just drop the relation from the
+		 * publication.
+		 */
+		if (j == 0)
+		{
+			ereport(ERROR,
+					errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					errmsg("cannot drop the last column in publication \"%s\"",
+						   get_publication_name(pubrel->prpubid, false)),
+					errhint("Remove table \"%s\" from the publication first.",
+							get_rel_name(pubrel->prrelid)));
+		}
+
+		/* Build the updated tuple */
+		MemSet(values, 0, sizeof(values));
+		MemSet(nulls, false, sizeof(nulls));
+		MemSet(replace, false, sizeof(replace));
+		newvec = buildint2vector(newelems, j);
+		values[Anum_pg_publication_rel_prattrs - 1] = PointerGetDatum(newvec);
+		replace[Anum_pg_publication_rel_prattrs - 1] = true;
+
+		/* Execute the update */
+		newtup = heap_modify_tuple(tup, RelationGetDescr(rel),
+								   values, nulls, replace);
+		CatalogTupleUpdate(rel, &tup->t_self, newtup);
+	}
 
 	ReleaseSysCache(tup);
 
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index c821271306..705bddc773 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -40,8 +40,9 @@
 #include "catalog/pg_inherits.h"
 #include "catalog/pg_namespace.h"
 #include "catalog/pg_opclass.h"
-#include "catalog/pg_tablespace.h"
+#include "catalog/pg_publication_rel.h"
 #include "catalog/pg_statistic_ext.h"
+#include "catalog/pg_tablespace.h"
 #include "catalog/pg_trigger.h"
 #include "catalog/pg_type.h"
 #include "catalog/storage.h"
@@ -8415,6 +8416,13 @@ ATExecDropColumn(List **wqueue, Relation rel, const char *colName,
 
 	ReleaseSysCache(tuple);
 
+	/*
+	 * If the column is part of a replication column list, arrange to get that
+	 * removed too.
+	 */
+	findAndAddAddresses(addrs, PublicationRelRelationId,
+						RelationRelationId, RelationGetRelid(rel), attnum);
+
 	/*
 	 * Propagate to children as appropriate.  Unlike most other ALTER
 	 * routines, we have to do this one level of recursion at a time; we can't
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index dfb5d0430c..f9f9ecd0c0 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -621,6 +621,8 @@ send_relation_and_attrs(Relation relation, TransactionId xid,
 		 * filter. XXX Allow sending type information for REPLICA IDENTITY
 		 * COLUMNS with user created type. even when they are not mentioned in
 		 * column filters.
+		 *
+		 * FIXME -- this code seems not verified by tests.
 		 */
 		if (att_map != NULL &&
 			!bms_is_member(att->attnum - FirstLowInvalidHeapAttributeNumber,
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 2f412ca3db..84ee807e0b 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1648,6 +1648,8 @@ psql_completion(const char *text, int start, int end)
 	/* ALTER PUBLICATION <name> ADD */
 	else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD"))
 		COMPLETE_WITH("ALL TABLES IN SCHEMA", "TABLE");
+	else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD", "TABLE"))
+		COMPLETE_WITH_SCHEMA_QUERY(Query_for_list_of_tables, NULL);
 	/* ALTER PUBLICATION <name> DROP */
 	else if (Matches("ALTER", "PUBLICATION", MatchAny, "DROP"))
 		COMPLETE_WITH("ALL TABLES IN SCHEMA", "TABLE");
diff --git a/src/include/catalog/dependency.h b/src/include/catalog/dependency.h
index 3eca295ff4..76d421e09e 100644
--- a/src/include/catalog/dependency.h
+++ b/src/include/catalog/dependency.h
@@ -214,6 +214,9 @@ extern long changeDependenciesOf(Oid classId, Oid oldObjectId,
 extern long changeDependenciesOn(Oid refClassId, Oid oldRefObjectId,
 								 Oid newRefObjectId);
 
+extern void findAndAddAddresses(ObjectAddresses *addrs, Oid classId,
+					Oid refclassId, Oid refobjectId, int32 refobjsubId);
+
 extern Oid	getExtensionOfObject(Oid classId, Oid objectId);
 extern List *getAutoExtensionsOfObject(Oid classId, Oid objectId);
 
diff --git a/src/include/commands/publicationcmds.h b/src/include/commands/publicationcmds.h
index 4ba68c70ee..23f037df7f 100644
--- a/src/include/commands/publicationcmds.h
+++ b/src/include/commands/publicationcmds.h
@@ -25,7 +25,7 @@
 extern ObjectAddress CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt);
 extern void AlterPublication(ParseState *pstate, AlterPublicationStmt *stmt);
 extern void RemovePublicationById(Oid pubid);
-extern void RemovePublicationRelById(Oid proid);
+extern void RemovePublicationRelById(Oid proid, int32 attnum);
 extern void RemovePublicationSchemaById(Oid psoid);
 
 extern ObjectAddress AlterPublicationOwner(const char *name, Oid newOwnerId);
diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
index 93ef6e21eb..aef2f905a1 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -167,18 +167,32 @@ Publications:
 
 CREATE TABLE testpub_tbl5 (a int PRIMARY KEY, b text, c text);
 ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (x, y, z);  -- error
-ERROR:  cannot add relation "testpub_tbl5" to publication
-DETAIL:  Column filter must include REPLICA IDENTITY columns
-ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (a, x);  -- error FIXME
+ERROR:  column "x" of relation "testpub_tbl5" does not exist
+ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (a, x);  -- error
+ERROR:  column "x" of relation "testpub_tbl5" does not exist
 ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (b, c);  -- error
-ERROR:  relation "testpub_tbl5" is already member of publication "testpub_fortable"
+ERROR:  invalid column list for publishing relation "testpub_tbl5"
+DETAIL:  All columns in REPLICA IDENTITY must be present in the column list.
 ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (a, c);  -- ok
-ERROR:  relation "testpub_tbl5" is already member of publication "testpub_fortable"
+ALTER TABLE testpub_tbl5 DROP COLUMN c;
+\dRp+ testpub_fortable
+                                Publication testpub_fortable
+          Owner           | All tables | Inserts | Updates | Deletes | Truncates | Via root 
+--------------------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f          | t       | t       | t       | t         | f
+Tables:
+    "public.testpub_tbl5" (a)
+Tables from schemas:
+    "pub_test"
+
+ALTER TABLE testpub_tbl5 DROP COLUMN a;
+ERROR:  cannot drop the last column in publication "testpub_fortable"
+HINT:  Remove table "testpub_tbl5" from the publication first.
 CREATE TABLE testpub_tbl6 (a int, b text, c text);
 ALTER TABLE testpub_tbl6 REPLICA IDENTITY FULL;
 ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl6 (a, b, c);  -- error
-ERROR:  cannot add relation "testpub_tbl6" to publication
-DETAIL:  Cannot have column filter with REPLICA IDENTITY FULL
+ERROR:  invalid column list for publishing relation "testpub_tbl6"
+DETAIL:  Cannot have column filter on relations with REPLICA IDENTITY FULL.
 DROP TABLE testpub_tbl2, testpub_tbl5, testpub_tbl6;
 DROP PUBLICATION testpub_foralltables, testpub_fortable, testpub_forschema;
 CREATE TABLE testpub_tbl3 (a int);
diff --git a/src/test/regress/sql/publication.sql b/src/test/regress/sql/publication.sql
index eb0e71ea62..18b87803c0 100644
--- a/src/test/regress/sql/publication.sql
+++ b/src/test/regress/sql/publication.sql
@@ -91,9 +91,12 @@ SELECT pubname, puballtables FROM pg_publication WHERE pubname = 'testpub_forall
 
 CREATE TABLE testpub_tbl5 (a int PRIMARY KEY, b text, c text);
 ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (x, y, z);  -- error
-ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (a, x);  -- error FIXME
+ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (a, x);  -- error
 ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (b, c);  -- error
 ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (a, c);  -- ok
+ALTER TABLE testpub_tbl5 DROP COLUMN c;
+\dRp+ testpub_fortable
+ALTER TABLE testpub_tbl5 DROP COLUMN a;
 
 CREATE TABLE testpub_tbl6 (a int, b text, c text);
 ALTER TABLE testpub_tbl6 REPLICA IDENTITY FULL;
diff --git a/src/test/subscription/t/021_column_filter.pl b/src/test/subscription/t/021_column_filter.pl
index 27d0537621..354e6ac363 100644
--- a/src/test/subscription/t/021_column_filter.pl
+++ b/src/test/subscription/t/021_column_filter.pl
@@ -5,7 +5,7 @@ use strict;
 use warnings;
 use PostgreSQL::Test::Cluster;
 use PostgreSQL::Test::Utils;
-use Test::More tests => 8;
+use Test::More tests => 10;
 
 # setup
 
@@ -129,14 +129,23 @@ $node_publisher->safe_psql('postgres',
 	"UPDATE tab2 SET c = 5 where a = 1");
 is($result, qq(1|abc), 'update on column c is not replicated');
 
-# Test error conditions
+# Test behavior when a column is dropped
 $node_publisher->safe_psql('postgres',
 	"ALTER TABLE test_part DROP COLUMN b");
 $result = $node_publisher->safe_psql('postgres',
-	"select relname, prattrs from pg_publication_rel pb, pg_class pc where pb.prrelid = pc.oid;");
-is($result, qq(tab1|1 2
+	"select prrelid::regclass, prattrs from pg_publication_rel pb;");
+is($result,
+	q(tab1|1 2
+tab3|1 3
 tab2|1 2
-tab3|1 3), 'publication relation test_part removed');
+test_part|1), 'column test_part.b removed');
+
+$node_publisher->safe_psql('postgres',
+	"INSERT INTO test_part VALUES (3, '2021-12-13 12:13:14')");
+$node_publisher->wait_for_catchup('sub1');
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM test_part WHERE a = 3");
+is($result, "3|", 'only column a is replicated');
 
 $node_publisher->safe_psql('postgres', "CREATE TABLE tab4 (a int PRIMARY KEY, b int, c int, d int)");
 $node_subscriber->safe_psql('postgres', "CREATE TABLE tab4 (a int PRIMARY KEY, b int, d int)");