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

Author: Amit Kapila

doc/src/sgml/logical-replication.sgml

@@ -1119,6 +1119,11 @@ test_sub=# SELECT * FROM child ORDER BY a;
    of columns in the list is not preserved.
   </para>
 
+  <para>
+   Specifying a column list when the publication also publishes
+   <literal>FOR TABLES IN SCHEMA</literal> is not supported.
+  </para>
+
   <para>
    For partitioned tables, the publication parameter
    <literal>publish_via_partition_root</literal> determines which column list

doc/src/sgml/ref/alter_publication.sgml

@@ -52,9 +52,11 @@ ALTER PUBLICATION <replaceable class="parameter">name</replaceable> RENAME TO <r
    remove one or more tables/schemas from the publication.  Note that adding
    tables/schemas to a publication that is already subscribed to will require an
    <literal>ALTER SUBSCRIPTION ... REFRESH PUBLICATION</literal> action on the
-   subscribing side in order to become effective. Note also that the combination
-   of <literal>DROP</literal> with a <literal>WHERE</literal> clause is not
-   allowed.
+   subscribing side in order to become effective. Note also that
+   <literal>DROP TABLES IN SCHEMA</literal> will not drop any schema tables
+   that were specified using <literal>FOR TABLE</literal>/
+   <literal>ADD TABLE</literal>, and the combination of <literal>DROP</literal>
+   with a <literal>WHERE</literal> clause is not allowed.
   </para>
 
   <para>
@@ -82,11 +84,8 @@ ALTER PUBLICATION <replaceable class="parameter">name</replaceable> RENAME TO <r
   </para>
 
   <para>
-   Adding/Setting a table that is part of schema specified in
-   <literal>TABLES IN SCHEMA</literal>, adding/setting a schema to a
-   publication that already has a table that is part of the specified schema or
-   adding/setting a table to a publication that already has a table's schema as
-   part of the specified schema is not supported.
+   Adding/Setting any schema when the publication also publishes a table with a
+   column list, and vice versa is not supported.
   </para>
  </refsect1>
 

doc/src/sgml/ref/create_publication.sgml

@@ -102,18 +102,18 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
       materialized views, and regular views cannot be part of a publication.
      </para>
 
+     <para>
+      Specifying a column list when the publication also publishes
+      <literal>FOR TABLES IN SCHEMA</literal> is not supported.
+     </para>
+
      <para>
       When a partitioned table is added to a publication, all of its existing
       and future partitions are implicitly considered to be part of the
       publication.  So, even operations that are performed directly on a
       partition are also published via publications that its ancestors are
       part of.
      </para>
-
-     <para>
-      Specifying a table that is part of a schema specified by
-      <literal>FOR TABLES IN SCHEMA</literal> is not supported.
-     </para>
     </listitem>
    </varlistentry>
 
@@ -136,8 +136,8 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
      </para>
 
      <para>
-      Specifying a schema along with a table which belongs to the specified
-      schema using <literal>FOR TABLE</literal> is not supported.
+      Specifying a schema when the publication also publishes a table with a
+      column list is not supported.
      </para>
 
      <para>
@@ -273,6 +273,12 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
    system columns.
   </para>
 
+  <para>
+   The row filter on a table becomes redundant if
+   <literal>FOR TABLES IN SCHEMA</literal> is specified and the table
+   belongs to the referred schema.
+  </para>
+
   <para>
    For published partitioned tables, the row filter for each
    partition is taken from the published partitioned table if the

src/backend/catalog/pg_publication.c

@@ -1111,6 +1111,7 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
 		HeapTuple	pubtuple = NULL;
 		HeapTuple	rettuple;
 		Oid			relid = list_nth_oid(tables, funcctx->call_cntr);
+		Oid			schemaid = get_rel_namespace(relid);
 		Datum		values[NUM_PUBLICATION_TABLES_ELEM] = {0};
 		bool		nulls[NUM_PUBLICATION_TABLES_ELEM] = {0};
 
@@ -1122,9 +1123,17 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
 
 		values[0] = ObjectIdGetDatum(relid);
 
-		pubtuple = SearchSysCacheCopy2(PUBLICATIONRELMAP,
-									   ObjectIdGetDatum(relid),
-									   ObjectIdGetDatum(publication->oid));
+		/*
+		 * We don't consider row filters or column lists for FOR ALL TABLES or
+		 * FOR TABLES IN SCHEMA publications.
+		 */
+		if (!publication->alltables &&
+			!SearchSysCacheExists2(PUBLICATIONNAMESPACEMAP,
+								   ObjectIdGetDatum(schemaid),
+								   ObjectIdGetDatum(publication->oid)))
+			pubtuple = SearchSysCacheCopy2(PUBLICATIONRELMAP,
+										   ObjectIdGetDatum(relid),
+										   ObjectIdGetDatum(publication->oid));
 
 		if (HeapTupleIsValid(pubtuple))
 		{

src/backend/commands/publicationcmds.c

@@ -66,7 +66,6 @@ typedef struct rf_context
 	Oid			parentid;		/* relid of the parent relation */
 } rf_context;
 
-static List *OpenRelIdList(List *relids);
 static List *OpenTableList(List *tables);
 static void CloseTableList(List *rels);
 static void LockSchemaList(List *schemalist);
@@ -214,44 +213,6 @@ ObjectsInPublicationToOids(List *pubobjspec_list, ParseState *pstate,
 	}
 }
 
-/*
- * Check if any of the given relation's schema is a member of the given schema
- * list.
- */
-static void
-CheckObjSchemaNotAlreadyInPublication(List *rels, List *schemaidlist,
-									  PublicationObjSpecType checkobjtype)
-{
-	ListCell   *lc;
-
-	foreach(lc, rels)
-	{
-		PublicationRelInfo *pub_rel = (PublicationRelInfo *) lfirst(lc);
-		Relation	rel = pub_rel->relation;
-		Oid			relSchemaId = RelationGetNamespace(rel);
-
-		if (list_member_oid(schemaidlist, relSchemaId))
-		{
-			if (checkobjtype == PUBLICATIONOBJ_TABLES_IN_SCHEMA)
-				ereport(ERROR,
-						errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-						errmsg("cannot add schema \"%s\" to publication",
-							   get_namespace_name(relSchemaId)),
-						errdetail("Table \"%s\" in schema \"%s\" is already part of the publication, adding the same schema is not supported.",
-								  RelationGetRelationName(rel),
-								  get_namespace_name(relSchemaId)));
-			else if (checkobjtype == PUBLICATIONOBJ_TABLE)
-				ereport(ERROR,
-						errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-						errmsg("cannot add relation \"%s.%s\" to publication",
-							   get_namespace_name(relSchemaId),
-							   RelationGetRelationName(rel)),
-						errdetail("Table's schema \"%s\" is already part of the publication or part of the specified schema list.",
-								  get_namespace_name(relSchemaId)));
-		}
-	}
-}
-
 /*
  * Returns true if any of the columns used in the row filter WHERE expression is
  * not part of REPLICA IDENTITY, false otherwise.
@@ -721,7 +682,7 @@ TransformPubWhereClauses(List *tables, const char *queryString,
  */
 static void
 CheckPubRelationColumnList(List *tables, const char *queryString,
-						   bool pubviaroot)
+						   bool publish_schema, bool pubviaroot)
 {
 	ListCell   *lc;
 
@@ -732,6 +693,24 @@ CheckPubRelationColumnList(List *tables, const char *queryString,
 		if (pri->columns == NIL)
 			continue;
 
+		/*
+		 * Disallow specifying column list if any schema is in the
+		 * publication.
+		 *
+		 * XXX We could instead just forbid the case when the publication
+		 * tries to publish the table with a column list and a schema for that
+		 * table. However, if we do that then we need a restriction during
+		 * ALTER TABLE ... SET SCHEMA to prevent such a case which doesn't
+		 * seem to be a good idea.
+		 */
+		if (publish_schema)
+			ereport(ERROR,
+					errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					errmsg("cannot use publication column list for relation \"%s.%s\"",
+						   get_namespace_name(RelationGetNamespace(pri->relation)),
+						   RelationGetRelationName(pri->relation)),
+					errdetail("Column list cannot be specified if any schema is part of the publication or specified in the list."));
+
 		/*
 		 * If the publication doesn't publish changes via the root partitioned
 		 * table, the partition's column list will be used. So disallow using
@@ -858,13 +837,11 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
 			List	   *rels;
 
 			rels = OpenTableList(relations);
-			CheckObjSchemaNotAlreadyInPublication(rels, schemaidlist,
-												  PUBLICATIONOBJ_TABLE);
-
 			TransformPubWhereClauses(rels, pstate->p_sourcetext,
 									 publish_via_partition_root);
 
 			CheckPubRelationColumnList(rels, pstate->p_sourcetext,
+									   schemaidlist != NIL,
 									   publish_via_partition_root);
 
 			PublicationAddTables(puboid, rels, true, NULL);
@@ -1110,8 +1087,8 @@ InvalidatePublicationRels(List *relids)
  */
 static void
 AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup,
-					   List *tables, List *schemaidlist,
-					   const char *queryString)
+					   List *tables, const char *queryString,
+					   bool publish_schema)
 {
 	List	   *rels = NIL;
 	Form_pg_publication pubform = (Form_pg_publication) GETSTRUCT(tup);
@@ -1129,19 +1106,12 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup,
 
 	if (stmt->action == AP_AddObjects)
 	{
-		List	   *schemas = NIL;
-
-		/*
-		 * Check if the relation is member of the existing schema in the
-		 * publication or member of the schema list specified.
-		 */
-		schemas = list_concat_copy(schemaidlist, GetPublicationSchemas(pubid));
-		CheckObjSchemaNotAlreadyInPublication(rels, schemas,
-											  PUBLICATIONOBJ_TABLE);
-
 		TransformPubWhereClauses(rels, queryString, pubform->pubviaroot);
 
-		CheckPubRelationColumnList(rels, queryString, pubform->pubviaroot);
+		publish_schema |= is_schema_publication(pubid);
+
+		CheckPubRelationColumnList(rels, queryString, publish_schema,
+								   pubform->pubviaroot);
 
 		PublicationAddTables(pubid, rels, false, stmt);
 	}
@@ -1154,12 +1124,10 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup,
 		List	   *delrels = NIL;
 		ListCell   *oldlc;
 
-		CheckObjSchemaNotAlreadyInPublication(rels, schemaidlist,
-											  PUBLICATIONOBJ_TABLE);
-
 		TransformPubWhereClauses(rels, queryString, pubform->pubviaroot);
 
-		CheckPubRelationColumnList(rels, queryString, pubform->pubviaroot);
+		CheckPubRelationColumnList(rels, queryString, publish_schema,
+								   pubform->pubviaroot);
 
 		/*
 		 * To recreate the relation list for the publication, look for
@@ -1308,16 +1276,35 @@ AlterPublicationSchemas(AlterPublicationStmt *stmt,
 	LockSchemaList(schemaidlist);
 	if (stmt->action == AP_AddObjects)
 	{
-		List	   *rels;
+		ListCell   *lc;
 		List	   *reloids;
 
 		reloids = GetPublicationRelations(pubform->oid, PUBLICATION_PART_ROOT);
-		rels = OpenRelIdList(reloids);
 
-		CheckObjSchemaNotAlreadyInPublication(rels, schemaidlist,
-											  PUBLICATIONOBJ_TABLES_IN_SCHEMA);
+		foreach(lc, reloids)
+		{
+			HeapTuple	coltuple;
+
+			coltuple = SearchSysCache2(PUBLICATIONRELMAP,
+									   ObjectIdGetDatum(lfirst_oid(lc)),
+									   ObjectIdGetDatum(pubform->oid));
+
+			if (!HeapTupleIsValid(coltuple))
+				continue;
+
+			/*
+			 * Disallow adding schema if column list is already part of the
+			 * publication. See CheckPubRelationColumnList.
+			 */
+			if (!heap_attisnull(coltuple, Anum_pg_publication_rel_prattrs, NULL))
+				ereport(ERROR,
+						errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						errmsg("cannot add schema to the publication"),
+						errdetail("Schema cannot be added if any table that specifies column list is already part of the publication."));
+
+			ReleaseSysCache(coltuple);
+		}
 
-		CloseTableList(rels);
 		PublicationAddSchemas(pubform->oid, schemaidlist, false, stmt);
 	}
 	else if (stmt->action == AP_DropObjects)
@@ -1429,14 +1416,7 @@ AlterPublication(ParseState *pstate, AlterPublicationStmt *stmt)
 
 		heap_freetuple(tup);
 
-		/*
-		 * Lock the publication so nobody else can do anything with it. This
-		 * prevents concurrent alter to add table(s) that were already going
-		 * to become part of the publication by adding corresponding schema(s)
-		 * via this command and similarly it will prevent the concurrent
-		 * addition of schema(s) for which there is any corresponding table
-		 * being added by this command.
-		 */
+		/* Lock the publication so nobody else can do anything with it. */
 		LockDatabaseObject(PublicationRelationId, pubid, 0,
 						   AccessExclusiveLock);
 
@@ -1453,8 +1433,8 @@ AlterPublication(ParseState *pstate, AlterPublicationStmt *stmt)
 					errmsg("publication \"%s\" does not exist",
 						   stmt->pubname));
 
-		AlterPublicationTables(stmt, tup, relations, schemaidlist,
-							   pstate->p_sourcetext);
+		AlterPublicationTables(stmt, tup, relations, pstate->p_sourcetext,
+							   schemaidlist != NIL);
 		AlterPublicationSchemas(stmt, tup, schemaidlist);
 	}
 
@@ -1569,32 +1549,6 @@ RemovePublicationSchemaById(Oid psoid)
 	table_close(rel, RowExclusiveLock);
 }
 
-/*
- * Open relations specified by a relid list.
- * The returned tables are locked in ShareUpdateExclusiveLock mode in order to
- * add them to a publication.
- */
-static List *
-OpenRelIdList(List *relids)
-{
-	ListCell   *lc;
-	List	   *rels = NIL;
-
-	foreach(lc, relids)
-	{
-		PublicationRelInfo *pub_rel;
-		Oid			relid = lfirst_oid(lc);
-		Relation	rel = table_open(relid,
-									 ShareUpdateExclusiveLock);
-
-		pub_rel = palloc(sizeof(PublicationRelInfo));
-		pub_rel->relation = rel;
-		rels = lappend(rels, pub_rel);
-	}
-
-	return rels;
-}
-
 /*
  * Open relations specified by a PublicationTable list.
  * The returned tables are locked in ShareUpdateExclusiveLock mode in order to