Defend against unsupported partition relkind in logical replication worker.

Since partitions can be foreign tables not only plain tables, but
logical replication only supports plain tables, we'd better check the
relkind of a partition after we find it.  (There was some discussion
of checking this when adding a partitioned table to a subscription;
but that would be inadequate since the troublesome partition could be
added later.)  Without this, the situation leads to a segfault or
assertion failure.

In passing, add a separate variable for the target Relation of
a cross-partition UPDATE; reusing partrel seemed mighty confusing
and error-prone.

Shi Yu and Tom Lane, per report from Ilya Gladyshev.  Back-patch
to v13 where logical replication into partitioned tables became
a thing.

Discussion: https://postgr.es/m/6b93e3748ba43298694f376ca8797279d7945e29.camel@gmail.com

diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 5250ae7f54c072ef9a17317230036a8da7aea2f6..e48a3f589aefbfe4369c38f7fee1c976679ea186 100644 (file)
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -2176,6 +2176,15 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
        Assert(partrelinfo != NULL);
        partrel = partrelinfo->ri_RelationDesc;
 
+       /*
+        * Check for supported relkind.  We need this since partitions might be of
+        * unsupported relkinds; and the set of partitions can change, so checking
+        * at CREATE/ALTER SUBSCRIPTION would be insufficient.
+        */
+       CheckSubscriptionRelkind(partrel->rd_rel->relkind,
+                                                        get_namespace_name(RelationGetNamespace(partrel)),
+                                                        RelationGetRelationName(partrel));
+
        /*
         * To perform any of the operations below, the tuple must match the
         * partition's rowtype. Convert if needed or just copy, using a dedicated
@@ -2229,6 +2238,7 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
                        {
                                TupleTableSlot *localslot;
                                ResultRelInfo *partrelinfo_new;
+                               Relation        partrel_new;
                                bool            found;
 
                                /* Get the matching local tuple from the partition. */
@@ -2314,7 +2324,6 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
                                                slot_getallattrs(remoteslot);
                                        }
 
-
                                        /* Find the new partition. */
                                        oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
                                        partrelinfo_new = ExecFindPartition(mtstate, relinfo,
@@ -2322,6 +2331,12 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
                                                                                                                estate);
                                        MemoryContextSwitchTo(oldctx);
                                        Assert(partrelinfo_new != partrelinfo);
+                                       partrel_new = partrelinfo_new->ri_RelationDesc;
+
+                                       /* Check that new partition also has supported relkind. */
+                                       CheckSubscriptionRelkind(partrel_new->rd_rel->relkind,
+                                                                                        get_namespace_name(RelationGetNamespace(partrel_new)),
+                                                                                        RelationGetRelationName(partrel_new));
 
                                        /* DELETE old tuple found in the old partition. */
                                        apply_handle_delete_internal(edata, partrelinfo,
@@ -2334,10 +2349,9 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
                                         * partition rowtype.
                                         */
                                        oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
-                                       partrel = partrelinfo_new->ri_RelationDesc;
                                        remoteslot_part = partrelinfo_new->ri_PartitionTupleSlot;
                                        if (remoteslot_part == NULL)
-                                               remoteslot_part = table_slot_create(partrel,
+                                               remoteslot_part = table_slot_create(partrel_new,
                                                                                                                        &estate->es_tupleTable);
                                        map = partrelinfo_new->ri_RootToPartitionMap;
                                        if (map != NULL)