Re: Subscription tests fail under CLOBBER_CACHE_ALWAYS

Amit Langote <amitlangote09@gmail.com> writes:
> IOW, the patch you posted earlier seems like the way to go.

I really dislike that patch.  I think it's doubling down on the messy,
unstructured coding patterns that got us into this situation to begin
with.  I'd prefer to expend a little effort on refactoring so that
the ExecCleanupTupleRouting call can be moved to the cleanup function
where it belongs.

So, I propose the attached, which invents a new struct to carry
the stuff we've discovered to be necessary.  This makes the APIs
noticeably cleaner IMHO.

I did not touch the APIs of the apply_XXX_internal functions,
as it didn't really seem to offer any notational advantage.
We can't simply collapse them to take an "edata" as I did for
apply_handle_tuple_routing, because the ResultRelInfo they're
supposed to operate on could be different from the original one.
I considered a couple of alternatives:

* Replace their estate arguments with edata, but keep the separate
ResultRelInfo arguments.  This might be worth doing in future, if we
add more fields to ApplyExecutionData.  Right now it'd save nothing,
and it'd create a risk of confusion about when to use the
ResultRelInfo argument vs. edata->resultRelInfo.

* Allow apply_handle_tuple_routing to overwrite edata->resultRelInfo
with the partition child's RRI, then simplify the apply_XXX_internal
functions to take just edata instead of separate estate and
resultRelInfo args.  I think this would work right now, but it seems
grotty, and it might cause problems in future.

* Replace the edata->resultRelInfo field with two fields, one for
the original parent and one for the actual/current target.  Perhaps
this is worth doing, not sure.

Thoughts?

            regards, tom lane

diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 1432554d5a..c51a83f797 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -178,6 +178,14 @@ static HTAB *xidhash = NULL;
 /* BufFile handle of the current streaming file */
 static BufFile *stream_fd = NULL;

+typedef struct ApplyExecutionData
+{
+    EState       *estate;            /* always needed */
+    ResultRelInfo *resultRelInfo;    /* always needed */
+    ModifyTableState *mtstate;    /* used for partitioned target rel */
+    PartitionTupleRouting *proute;    /* used for partitioned target rel */
+} ApplyExecutionData;
+
 typedef struct SubXactInfo
 {
     TransactionId xid;            /* XID of the subxact */
@@ -239,8 +247,7 @@ static bool FindReplTupleInLocalRel(EState *estate, Relation localrel,
                                     LogicalRepRelation *remoterel,
                                     TupleTableSlot *remoteslot,
                                     TupleTableSlot **localslot);
-static void apply_handle_tuple_routing(ResultRelInfo *relinfo,
-                                       EState *estate,
+static void apply_handle_tuple_routing(ApplyExecutionData *edata,
                                        TupleTableSlot *remoteslot,
                                        LogicalRepTupleData *newtup,
                                        LogicalRepRelMapEntry *relmapentry,
@@ -336,20 +343,21 @@ handle_streamed_transaction(LogicalRepMsgType action, StringInfo s)

 /*
  * Executor state preparation for evaluation of constraint expressions,
- * indexes and triggers.
+ * indexes and triggers for the specified relation.
  *
- * resultRelInfo is a ResultRelInfo for the relation to be passed to the
- * executor routines.  The caller must open and close any indexes to be
- * updated independently of the relation registered here.
+ * Note that the caller must open and close any indexes to be updated.
  */
-static EState *
-create_estate_for_relation(LogicalRepRelMapEntry *rel,
-                           ResultRelInfo **resultRelInfo)
+static ApplyExecutionData *
+create_edata_for_relation(LogicalRepRelMapEntry *rel)
 {
+    ApplyExecutionData *edata;
     EState       *estate;
     RangeTblEntry *rte;
+    ResultRelInfo *resultRelInfo;

-    estate = CreateExecutorState();
+    edata = (ApplyExecutionData *) palloc0(sizeof(ApplyExecutionData));
+
+    edata->estate = estate = CreateExecutorState();

     rte = makeNode(RangeTblEntry);
     rte->rtekind = RTE_RELATION;
@@ -358,13 +366,13 @@ create_estate_for_relation(LogicalRepRelMapEntry *rel,
     rte->rellockmode = AccessShareLock;
     ExecInitRangeTable(estate, list_make1(rte));

-    *resultRelInfo = makeNode(ResultRelInfo);
+    edata->resultRelInfo = resultRelInfo = makeNode(ResultRelInfo);

     /*
      * Use Relation opened by logicalrep_rel_open() instead of opening it
      * again.
      */
-    InitResultRelInfo(*resultRelInfo, rel->localrel, 1, NULL, 0);
+    InitResultRelInfo(resultRelInfo, rel->localrel, 1, NULL, 0);

     /*
      * We put the ResultRelInfo in the es_opened_result_relations list, even
@@ -377,29 +385,38 @@ create_estate_for_relation(LogicalRepRelMapEntry *rel,
      * an apply operation being responsible for that.
      */
     estate->es_opened_result_relations =
-        lappend(estate->es_opened_result_relations, *resultRelInfo);
+        lappend(estate->es_opened_result_relations, resultRelInfo);

     estate->es_output_cid = GetCurrentCommandId(true);

     /* Prepare to catch AFTER triggers. */
     AfterTriggerBeginQuery();

-    return estate;
+    /* other fields of edata remain NULL for now */
+
+    return edata;
 }

 /*
  * Finish any operations related to the executor state created by
- * create_estate_for_relation().
+ * create_edata_for_relation().
  */
 static void
-finish_estate(EState *estate)
+finish_edata(ApplyExecutionData *edata)
 {
+    EState       *estate = edata->estate;
+
     /* Handle any queued AFTER triggers. */
     AfterTriggerEndQuery(estate);

+    /* Shut down tuple routing, if any was done. */
+    if (edata->proute)
+        ExecCleanupTupleRouting(edata->mtstate, edata->proute);
+
     /* Cleanup. */
     ExecResetTupleTable(estate->es_tupleTable, false);
     FreeExecutorState(estate);
+    pfree(edata);
 }

 /*
@@ -1181,10 +1198,10 @@ GetRelationIdentityOrPK(Relation rel)
 static void
 apply_handle_insert(StringInfo s)
 {
-    ResultRelInfo *resultRelInfo;
     LogicalRepRelMapEntry *rel;
     LogicalRepTupleData newtup;
     LogicalRepRelId relid;
+    ApplyExecutionData *edata;
     EState       *estate;
     TupleTableSlot *remoteslot;
     MemoryContext oldctx;
@@ -1207,7 +1224,8 @@ apply_handle_insert(StringInfo s)
     }

     /* Initialize the executor state. */
-    estate = create_estate_for_relation(rel, &resultRelInfo);
+    edata = create_edata_for_relation(rel);
+    estate = edata->estate;
     remoteslot = ExecInitExtraTupleSlot(estate,
                                         RelationGetDescr(rel->localrel),
                                         &TTSOpsVirtual);
@@ -1223,15 +1241,15 @@ apply_handle_insert(StringInfo s)

     /* For a partitioned table, insert the tuple into a partition. */
     if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
-        apply_handle_tuple_routing(resultRelInfo, estate,
+        apply_handle_tuple_routing(edata,
                                    remoteslot, NULL, rel, CMD_INSERT);
     else
-        apply_handle_insert_internal(resultRelInfo, estate,
+        apply_handle_insert_internal(edata->resultRelInfo, estate,
                                      remoteslot);

     PopActiveSnapshot();

-    finish_estate(estate);
+    finish_edata(edata);

     logicalrep_rel_close(rel, NoLock);

@@ -1293,9 +1311,9 @@ check_relation_updatable(LogicalRepRelMapEntry *rel)
 static void
 apply_handle_update(StringInfo s)
 {
-    ResultRelInfo *resultRelInfo;
     LogicalRepRelMapEntry *rel;
     LogicalRepRelId relid;
+    ApplyExecutionData *edata;
     EState       *estate;
     LogicalRepTupleData oldtup;
     LogicalRepTupleData newtup;
@@ -1326,7 +1344,8 @@ apply_handle_update(StringInfo s)
     check_relation_updatable(rel);

     /* Initialize the executor state. */
-    estate = create_estate_for_relation(rel, &resultRelInfo);
+    edata = create_edata_for_relation(rel);
+    estate = edata->estate;
     remoteslot = ExecInitExtraTupleSlot(estate,
                                         RelationGetDescr(rel->localrel),
                                         &TTSOpsVirtual);
@@ -1368,15 +1387,15 @@ apply_handle_update(StringInfo s)

     /* For a partitioned table, apply update to correct partition. */
     if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
-        apply_handle_tuple_routing(resultRelInfo, estate,
+        apply_handle_tuple_routing(edata,
                                    remoteslot, &newtup, rel, CMD_UPDATE);
     else
-        apply_handle_update_internal(resultRelInfo, estate,
+        apply_handle_update_internal(edata->resultRelInfo, estate,
                                      remoteslot, &newtup, rel);

     PopActiveSnapshot();

-    finish_estate(estate);
+    finish_edata(edata);

     logicalrep_rel_close(rel, NoLock);

@@ -1448,10 +1467,10 @@ apply_handle_update_internal(ResultRelInfo *relinfo,
 static void
 apply_handle_delete(StringInfo s)
 {
-    ResultRelInfo *resultRelInfo;
     LogicalRepRelMapEntry *rel;
     LogicalRepTupleData oldtup;
     LogicalRepRelId relid;
+    ApplyExecutionData *edata;
     EState       *estate;
     TupleTableSlot *remoteslot;
     MemoryContext oldctx;
@@ -1477,7 +1496,8 @@ apply_handle_delete(StringInfo s)
     check_relation_updatable(rel);

     /* Initialize the executor state. */
-    estate = create_estate_for_relation(rel, &resultRelInfo);
+    edata = create_edata_for_relation(rel);
+    estate = edata->estate;
     remoteslot = ExecInitExtraTupleSlot(estate,
                                         RelationGetDescr(rel->localrel),
                                         &TTSOpsVirtual);
@@ -1491,15 +1511,15 @@ apply_handle_delete(StringInfo s)

     /* For a partitioned table, apply delete to correct partition. */
     if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
-        apply_handle_tuple_routing(resultRelInfo, estate,
+        apply_handle_tuple_routing(edata,
                                    remoteslot, NULL, rel, CMD_DELETE);
     else
-        apply_handle_delete_internal(resultRelInfo, estate,
+        apply_handle_delete_internal(edata->resultRelInfo, estate,
                                      remoteslot, &rel->remoterel);

     PopActiveSnapshot();

-    finish_estate(estate);
+    finish_edata(edata);

     logicalrep_rel_close(rel, NoLock);

@@ -1582,16 +1602,17 @@ FindReplTupleInLocalRel(EState *estate, Relation localrel,
  * This handles insert, update, delete on a partitioned table.
  */
 static void
-apply_handle_tuple_routing(ResultRelInfo *relinfo,
-                           EState *estate,
+apply_handle_tuple_routing(ApplyExecutionData *edata,
                            TupleTableSlot *remoteslot,
                            LogicalRepTupleData *newtup,
                            LogicalRepRelMapEntry *relmapentry,
                            CmdType operation)
 {
+    EState       *estate = edata->estate;
+    ResultRelInfo *relinfo = edata->resultRelInfo;
     Relation    parentrel = relinfo->ri_RelationDesc;
-    ModifyTableState *mtstate = NULL;
-    PartitionTupleRouting *proute = NULL;
+    ModifyTableState *mtstate;
+    PartitionTupleRouting *proute;
     ResultRelInfo *partrelinfo;
     Relation    partrel;
     TupleTableSlot *remoteslot_part;
@@ -1599,12 +1620,14 @@ apply_handle_tuple_routing(ResultRelInfo *relinfo,
     MemoryContext oldctx;

     /* ModifyTableState is needed for ExecFindPartition(). */
-    mtstate = makeNode(ModifyTableState);
+    edata->mtstate = mtstate = makeNode(ModifyTableState);
     mtstate->ps.plan = NULL;
     mtstate->ps.state = estate;
     mtstate->operation = operation;
     mtstate->resultRelInfo = relinfo;
-    proute = ExecSetupPartitionTupleRouting(estate, parentrel);
+
+    /* ... as is PartitionTupleRouting. */
+    edata->proute = proute = ExecSetupPartitionTupleRouting(estate, parentrel);

     /*
      * Find the partition to which the "search tuple" belongs.
@@ -1797,8 +1820,6 @@ apply_handle_tuple_routing(ResultRelInfo *relinfo,
             elog(ERROR, "unrecognized CmdType: %d", (int) operation);
             break;
     }
-
-    ExecCleanupTupleRouting(mtstate, proute);
 }

 /*