Allow publications with schema and table of the same schema.

We previously thought that allowing such cases can confuse users when they
specify DROP TABLES IN SCHEMA but that doesn't seem to be the case based
on discussion. This helps to uplift the restriction during
ALTER TABLE ... SET SCHEMA which used to ensure that we couldn't end up
with a publication having both a schema and the same schema's table.

To allow this, we need to forbid having any schema on a publication if
column lists on a table are specified (and vice versa). This is because
otherwise we still need a restriction during ALTER TABLE ... SET SCHEMA to
forbid cases where it could lead to a publication having both a schema and
the same schema's table with column list.

Based on suggestions by Peter Eisentraut.

Author: Hou Zhijie and Vignesh C
Reviewed-By: Peter Smith, Amit Kapila
Backpatch-through: 15, where it was introduced
Discussion: https://postgr.es/m/2729c9e2-9aac-8cda-f2f4-34f2bcc18f4e@enterprisedb.com

1 files changed, 14 insertions, 17 deletions

diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 92bbffbe7cd..2ecaa5b9074 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -854,6 +854,7 @@ pgoutput_row_filter_init(PGOutputData *data, List *publications,
 	MemoryContext oldctx;
 	int			idx;
 	bool		has_filter = true;
+	Oid			schemaid = get_rel_namespace(entry->publish_as_relid);
 
 	/*
 	 * Find if there are any row filters for this relation. If there are, then
@@ -867,26 +868,26 @@ pgoutput_row_filter_init(PGOutputData *data, List *publications,
 	 * are multiple lists (one for each operation) to which row filters will
 	 * be appended.
 	 *
-	 * FOR ALL TABLES implies "don't use row filter expression" so it takes
-	 * precedence.
+	 * FOR ALL TABLES and FOR TABLES IN SCHEMA implies "don't use row
+	 * filter expression" so it takes precedence.
 	 */
 	foreach(lc, publications)
 	{
 		Publication *pub = lfirst(lc);
 		HeapTuple	rftuple = NULL;
 		Datum		rfdatum = 0;
-		bool		pub_no_filter = false;
+		bool		pub_no_filter = true;
 
-		if (pub->alltables)
-		{
-			/*
-			 * If the publication is FOR ALL TABLES then it is treated the
-			 * same as if this table has no row filters (even if for other
-			 * publications it does).
-			 */
-			pub_no_filter = true;
-		}
-		else
+		/*
+		 * If the publication is FOR ALL TABLES, or the publication includes a
+		 * FOR TABLES IN SCHEMA where the table belongs to the referred
+		 * schema, then it is treated the same as if there are no row filters
+		 * (even if other publications have a row filter).
+		 */
+		if (!pub->alltables &&
+			!SearchSysCacheExists2(PUBLICATIONNAMESPACEMAP,
+								   ObjectIdGetDatum(schemaid),
+								   ObjectIdGetDatum(pub->oid)))
 		{
 			/*
 			 * Check for the presence of a row filter in this publication.
@@ -902,10 +903,6 @@ pgoutput_row_filter_init(PGOutputData *data, List *publications,
 										  Anum_pg_publication_rel_prqual,
 										  &pub_no_filter);
 			}
-			else
-			{
-				pub_no_filter = true;
-			}
 		}
 
 		if (pub_no_filter)