Allow publishing partition changes via ancestors

To control whether partition changes are replicated using their own
identity and schema or an ancestor's, add a new parameter that can be
set per publication named 'publish_via_partition_root'.

This allows replicating a partitioned table into a different partition
structure on the subscriber.

Author: Amit Langote <[email protected]>
Reviewed-by: Rafia Sabih <[email protected]>
Reviewed-by: Peter Eisentraut <[email protected]>
Reviewed-by: Petr Jelinek <[email protected]>
Discussion: https://www.postgresql.org/message-id/flat/CA+HiwqH=Y85vRK3mOdjEkqFK+E=ST=eQiHdpj43L=_eJMOOznQ@mail.gmail.com

Author: petere

Diff for: doc/src/sgml/catalogs.sgml

@@ -5437,6 +5437,16 @@ SCRAM-SHA-256$<replaceable>&lt;iteration count&gt;</replaceable>:<replaceable>&l
       <entry>If true, <command>TRUNCATE</command> operations are replicated for
        tables in the publication.</entry>
      </row>
+
+     <row>
+      <entry><structfield>pubviaroot</structfield></entry>
+      <entry><type>bool</type></entry>
+      <entry></entry>
+      <entry>If true, operations on a leaf partition are replicated using the
+       identity and schema of its topmost partitioned ancestor mentioned in the
+       publication instead of its own.
+      </entry>
+     </row>
     </tbody>
    </tgroup>
   </table>

Diff for: doc/src/sgml/logical-replication.sgml

@@ -411,10 +411,14 @@
    <listitem>
     <para>
      When replicating between partitioned tables, the actual replication
-     originates from the leaf partitions on the publisher, so partitions on
-     the publisher must also exist on the subscriber as valid target tables.
-     (They could either be leaf partitions themselves, or they could be
-     further subpartitioned, or they could even be independent tables.)
+     originates, by default, from the leaf partitions on the publisher, so
+     partitions on the publisher must also exist on the subscriber as valid
+     target tables. (They could either be leaf partitions themselves, or they
+     could be further subpartitioned, or they could even be independent
+     tables.)  Publications can also specify that changes are to be replicated
+     using the identity and schema of the partitioned root table instead of
+     that of the individual leaf partitions in which the changes actually
+     originate (see <xref linkend="sql-createpublication"/>).
     </para>
    </listitem>
   </itemizedlist>

Diff for: doc/src/sgml/ref/create_publication.sgml

@@ -123,6 +123,26 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
          </para>
         </listitem>
        </varlistentry>
+
+       <varlistentry>
+        <term><literal>publish_via_partition_root</literal> (<type>boolean</type>)</term>
+        <listitem>
+         <para>
+          This parameter determines whether changes in a partitioned table (or
+          on its partitions) contained in the publication will be published
+          using the identity and schema of the partitioned table rather than
+          that of the individual partitions that are actually changed; the
+          latter is the default.  Enablings this allows the changes to be
+          replicated into a non-partitioned table or a partitioned table
+          consisting of a different set of partitions.
+         </para>
+
+         <para>
+          If this is enabled, <literal>TRUNCATE</literal> operations performed
+          directly on partitions are not replicated.
+         </para>
+        </listitem>
+       </varlistentry>
       </variablelist>
 
      </para>

Diff for: src/backend/catalog/pg_publication.c

@@ -42,8 +42,6 @@
 #include "utils/rel.h"
 #include "utils/syscache.h"
 
-static List *get_rel_publications(Oid relid);
-
 /*
  * Check if relation can be in given publication and throws appropriate
  * error if not.
@@ -216,37 +214,9 @@ publication_add_relation(Oid pubid, Relation targetrel,
 	return myself;
 }
 
-
-/*
- * Gets list of publication oids for a relation, plus those of ancestors,
- * if any, if the relation is a partition.
- */
+/* Gets list of publication oids for a relation */
 List *
 GetRelationPublications(Oid relid)
-{
-	List	   *result = NIL;
-
-	result = get_rel_publications(relid);
-	if (get_rel_relispartition(relid))
-	{
-		List	   *ancestors = get_partition_ancestors(relid);
-		ListCell   *lc;
-
-		foreach(lc, ancestors)
-		{
-			Oid			ancestor = lfirst_oid(lc);
-			List	   *ancestor_pubs = get_rel_publications(ancestor);
-
-			result = list_concat(result, ancestor_pubs);
-		}
-	}
-
-	return result;
-}
-
-/* Workhorse of GetRelationPublications() */
-static List *
-get_rel_publications(Oid relid)
 {
 	List	   *result = NIL;
 	CatCList   *pubrellist;
@@ -373,9 +343,13 @@ GetAllTablesPublications(void)
 
 /*
  * Gets list of all relation published by FOR ALL TABLES publication(s).
+ *
+ * If the publication publishes partition changes via their respective root
+ * partitioned tables, we must exclude partitions in favor of including the
+ * root partitioned tables.
  */
 List *
-GetAllTablesPublicationRelations(void)
+GetAllTablesPublicationRelations(bool pubviaroot)
 {
 	Relation	classRel;
 	ScanKeyData key[1];
@@ -397,12 +371,35 @@ GetAllTablesPublicationRelations(void)
 		Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
 		Oid			relid = relForm->oid;
 
-		if (is_publishable_class(relid, relForm))
+		if (is_publishable_class(relid, relForm) &&
+			!(relForm->relispartition && pubviaroot))
 			result = lappend_oid(result, relid);
 	}
 
 	table_endscan(scan);
-	table_close(classRel, AccessShareLock);
+
+	if (pubviaroot)
+	{
+		ScanKeyInit(&key[0],
+					Anum_pg_class_relkind,
+					BTEqualStrategyNumber, F_CHAREQ,
+					CharGetDatum(RELKIND_PARTITIONED_TABLE));
+
+		scan = table_beginscan_catalog(classRel, 1, key);
+
+		while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
+		{
+			Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
+			Oid			relid = relForm->oid;
+
+			if (is_publishable_class(relid, relForm) &&
+				!relForm->relispartition)
+				result = lappend_oid(result, relid);
+		}
+
+		table_endscan(scan);
+		table_close(classRel, AccessShareLock);
+	}
 
 	return result;
 }
@@ -433,6 +430,7 @@ GetPublication(Oid pubid)
 	pub->pubactions.pubupdate = pubform->pubupdate;
 	pub->pubactions.pubdelete = pubform->pubdelete;
 	pub->pubactions.pubtruncate = pubform->pubtruncate;
+	pub->pubviaroot = pubform->pubviaroot;
 
 	ReleaseSysCache(tup);
 
@@ -533,9 +531,11 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
 		 * need those.
 		 */
 		if (publication->alltables)
-			tables = GetAllTablesPublicationRelations();
+			tables = GetAllTablesPublicationRelations(publication->pubviaroot);
 		else
 			tables = GetPublicationRelations(publication->oid,
+											 publication->pubviaroot ?
+											 PUBLICATION_PART_ROOT :
 											 PUBLICATION_PART_LEAF);
 		funcctx->user_fctx = (void *) tables;
 

Diff for: src/backend/commands/publicationcmds.c

@@ -23,6 +23,7 @@
 #include "catalog/namespace.h"
 #include "catalog/objectaccess.h"
 #include "catalog/objectaddress.h"
+#include "catalog/partition.h"
 #include "catalog/pg_inherits.h"
 #include "catalog/pg_publication.h"
 #include "catalog/pg_publication_rel.h"
@@ -56,20 +57,21 @@ static void PublicationDropTables(Oid pubid, List *rels, bool missing_ok);
 static void
 parse_publication_options(List *options,
 						  bool *publish_given,
-						  bool *publish_insert,
-						  bool *publish_update,
-						  bool *publish_delete,
-						  bool *publish_truncate)
+						  PublicationActions *pubactions,
+						  bool *publish_via_partition_root_given,
+						  bool *publish_via_partition_root)
 {
 	ListCell   *lc;
 
 	*publish_given = false;
+	*publish_via_partition_root_given = false;
 
-	/* Defaults are true */
-	*publish_insert = true;
-	*publish_update = true;
-	*publish_delete = true;
-	*publish_truncate = true;
+	/* defaults */
+	pubactions->pubinsert = true;
+	pubactions->pubupdate = true;
+	pubactions->pubdelete = true;
+	pubactions->pubtruncate = true;
+	*publish_via_partition_root = false;
 
 	/* Parse options */
 	foreach(lc, options)
@@ -91,10 +93,10 @@ parse_publication_options(List *options,
 			 * If publish option was given only the explicitly listed actions
 			 * should be published.
 			 */
-			*publish_insert = false;
-			*publish_update = false;
-			*publish_delete = false;
-			*publish_truncate = false;
+			pubactions->pubinsert = false;
+			pubactions->pubupdate = false;
+			pubactions->pubdelete = false;
+			pubactions->pubtruncate = false;
 
 			*publish_given = true;
 			publish = defGetString(defel);
@@ -110,19 +112,28 @@ parse_publication_options(List *options,
 				char	   *publish_opt = (char *) lfirst(lc);
 
 				if (strcmp(publish_opt, "insert") == 0)
-					*publish_insert = true;
+					pubactions->pubinsert = true;
 				else if (strcmp(publish_opt, "update") == 0)
-					*publish_update = true;
+					pubactions->pubupdate = true;
 				else if (strcmp(publish_opt, "delete") == 0)
-					*publish_delete = true;
+					pubactions->pubdelete = true;
 				else if (strcmp(publish_opt, "truncate") == 0)
-					*publish_truncate = true;
+					pubactions->pubtruncate = true;
 				else
 					ereport(ERROR,
 							(errcode(ERRCODE_SYNTAX_ERROR),
 							 errmsg("unrecognized \"publish\" value: \"%s\"", publish_opt)));
 			}
 		}
+		else if (strcmp(defel->defname, "publish_via_partition_root") == 0)
+		{
+			if (*publish_via_partition_root_given)
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("conflicting or redundant options")));
+			*publish_via_partition_root_given = true;
+			*publish_via_partition_root = defGetBoolean(defel);
+		}
 		else
 			ereport(ERROR,
 					(errcode(ERRCODE_SYNTAX_ERROR),
@@ -143,10 +154,9 @@ CreatePublication(CreatePublicationStmt *stmt)
 	Datum		values[Natts_pg_publication];
 	HeapTuple	tup;
 	bool		publish_given;
-	bool		publish_insert;
-	bool		publish_update;
-	bool		publish_delete;
-	bool		publish_truncate;
+	PublicationActions pubactions;
+	bool		publish_via_partition_root_given;
+	bool		publish_via_partition_root;
 	AclResult	aclresult;
 
 	/* must have CREATE privilege on database */
@@ -183,23 +193,25 @@ CreatePublication(CreatePublicationStmt *stmt)
 	values[Anum_pg_publication_pubowner - 1] = ObjectIdGetDatum(GetUserId());
 
 	parse_publication_options(stmt->options,
-							  &publish_given, &publish_insert,
-							  &publish_update, &publish_delete,
-							  &publish_truncate);
+							  &publish_given, &pubactions,
+							  &publish_via_partition_root_given,
+							  &publish_via_partition_root);
 
 	puboid = GetNewOidWithIndex(rel, PublicationObjectIndexId,
 								Anum_pg_publication_oid);
 	values[Anum_pg_publication_oid - 1] = ObjectIdGetDatum(puboid);
 	values[Anum_pg_publication_puballtables - 1] =
 		BoolGetDatum(stmt->for_all_tables);
 	values[Anum_pg_publication_pubinsert - 1] =
-		BoolGetDatum(publish_insert);
+		BoolGetDatum(pubactions.pubinsert);
 	values[Anum_pg_publication_pubupdate - 1] =
-		BoolGetDatum(publish_update);
+		BoolGetDatum(pubactions.pubupdate);
 	values[Anum_pg_publication_pubdelete - 1] =
-		BoolGetDatum(publish_delete);
+		BoolGetDatum(pubactions.pubdelete);
 	values[Anum_pg_publication_pubtruncate - 1] =
-		BoolGetDatum(publish_truncate);
+		BoolGetDatum(pubactions.pubtruncate);
+	values[Anum_pg_publication_pubviaroot - 1] =
+		BoolGetDatum(publish_via_partition_root);
 
 	tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
 
@@ -251,17 +263,16 @@ AlterPublicationOptions(AlterPublicationStmt *stmt, Relation rel,
 	bool		replaces[Natts_pg_publication];
 	Datum		values[Natts_pg_publication];
 	bool		publish_given;
-	bool		publish_insert;
-	bool		publish_update;
-	bool		publish_delete;
-	bool		publish_truncate;
+	PublicationActions pubactions;
+	bool		publish_via_partition_root_given;
+	bool		publish_via_partition_root;
 	ObjectAddress obj;
 	Form_pg_publication pubform;
 
 	parse_publication_options(stmt->options,
-							  &publish_given, &publish_insert,
-							  &publish_update, &publish_delete,
-							  &publish_truncate);
+							  &publish_given, &pubactions,
+							  &publish_via_partition_root_given,
+							  &publish_via_partition_root);
 
 	/* Everything ok, form a new tuple. */
 	memset(values, 0, sizeof(values));
@@ -270,19 +281,25 @@ AlterPublicationOptions(AlterPublicationStmt *stmt, Relation rel,
 
 	if (publish_given)
 	{
-		values[Anum_pg_publication_pubinsert - 1] = BoolGetDatum(publish_insert);
+		values[Anum_pg_publication_pubinsert - 1] = BoolGetDatum(pubactions.pubinsert);
 		replaces[Anum_pg_publication_pubinsert - 1] = true;
 
-		values[Anum_pg_publication_pubupdate - 1] = BoolGetDatum(publish_update);
+		values[Anum_pg_publication_pubupdate - 1] = BoolGetDatum(pubactions.pubupdate);
 		replaces[Anum_pg_publication_pubupdate - 1] = true;
 
-		values[Anum_pg_publication_pubdelete - 1] = BoolGetDatum(publish_delete);
+		values[Anum_pg_publication_pubdelete - 1] = BoolGetDatum(pubactions.pubdelete);
 		replaces[Anum_pg_publication_pubdelete - 1] = true;
 
-		values[Anum_pg_publication_pubtruncate - 1] = BoolGetDatum(publish_truncate);
+		values[Anum_pg_publication_pubtruncate - 1] = BoolGetDatum(pubactions.pubtruncate);
 		replaces[Anum_pg_publication_pubtruncate - 1] = true;
 	}
 
+	if (publish_via_partition_root_given)
+	{
+		values[Anum_pg_publication_pubviaroot - 1] = BoolGetDatum(publish_via_partition_root);
+		replaces[Anum_pg_publication_pubviaroot - 1] = true;
+	}
+
 	tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
 							replaces);