Some refactoring of logical/worker.c

This moves the main operations of apply_handle_{insert|update|delete},
that of inserting, updating, deleting a tuple into/from a given
relation, into corresponding
apply_handle_{insert|update|delete}_internal functions.  This allows
performing those operations on relations that are not directly the
targets of replication, which is something a later patch will use for
targeting partitioned tables.

Author: Amit Langote <amitlangote09@gmail.com>
Reviewed-by: Rafia Sabih <rafia.pghackers@gmail.com>
Reviewed-by: Peter Eisentraut <peter.eisentraut@2ndquadrant.com>
Discussion: https://www.postgresql.org/message-id/flat/CA+HiwqH=Y85vRK3mOdjEkqFK+E=ST=eQiHdpj43L=_eJMOOznQ@mail.gmail.com

diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index ad4a732fd204440627e7610cd54a6bb00abf3204..e5f0aaa8f72594d3d05c1b5a00a8963fbb97727c 100644 (file)
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -113,6 +113,16 @@ static void store_flush_position(XLogRecPtr remote_lsn);
 
 static void maybe_reread_subscription(void);
 
+static void apply_handle_insert_internal(ResultRelInfo *relinfo,
+                                                                                EState *estate, TupleTableSlot *remoteslot);
+static void apply_handle_update_internal(ResultRelInfo *relinfo,
+                                                                                EState *estate, TupleTableSlot *remoteslot,
+                                                                                LogicalRepTupleData *newtup,
+                                                                                LogicalRepRelMapEntry *relmapentry);
+static void apply_handle_delete_internal(ResultRelInfo *relinfo, EState *estate,
+                                                                                TupleTableSlot *remoteslot,
+                                                                                LogicalRepRelation *remoterel);
+
 /*
  * Should this worker apply changes for given relation.
  *
@@ -582,6 +592,7 @@ GetRelationIdentityOrPK(Relation rel)
 /*
  * Handle INSERT message.
  */
+
 static void
 apply_handle_insert(StringInfo s)
 {
@@ -621,13 +632,10 @@ apply_handle_insert(StringInfo s)
        slot_fill_defaults(rel, estate, remoteslot);
        MemoryContextSwitchTo(oldctx);
 
-       ExecOpenIndices(estate->es_result_relation_info, false);
+       Assert(rel->localrel->rd_rel->relkind == RELKIND_RELATION);
+       apply_handle_insert_internal(estate->es_result_relation_info, estate,
+                                                                remoteslot);
 
-       /* Do the insert. */
-       ExecSimpleRelationInsert(estate, remoteslot);
-
-       /* Cleanup. */
-       ExecCloseIndices(estate->es_result_relation_info);
        PopActiveSnapshot();
 
        /* Handle queued AFTER triggers. */
@@ -641,6 +649,20 @@ apply_handle_insert(StringInfo s)
        CommandCounterIncrement();
 }
 
+/* Workhorse for apply_handle_insert() */
+static void
+apply_handle_insert_internal(ResultRelInfo *relinfo,
+                                                        EState *estate, TupleTableSlot *remoteslot)
+{
+       ExecOpenIndices(relinfo, false);
+
+       /* Do the insert. */
+       ExecSimpleRelationInsert(estate, remoteslot);
+
+       /* Cleanup. */
+       ExecCloseIndices(relinfo);
+}
+
 /*
  * Check if the logical replication relation is updatable and throw
  * appropriate error if it isn't.
@@ -684,16 +706,12 @@ apply_handle_update(StringInfo s)
 {
        LogicalRepRelMapEntry *rel;
        LogicalRepRelId relid;
-       Oid                     idxoid;
        EState     *estate;
-       EPQState        epqstate;
        LogicalRepTupleData oldtup;
        LogicalRepTupleData newtup;
        bool            has_oldtup;
-       TupleTableSlot *localslot;
        TupleTableSlot *remoteslot;
        RangeTblEntry *target_rte;
-       bool            found;
        MemoryContext oldctx;
 
        ensure_transaction();
@@ -719,9 +737,6 @@ apply_handle_update(StringInfo s)
        remoteslot = ExecInitExtraTupleSlot(estate,
                                                                                RelationGetDescr(rel->localrel),
                                                                                &TTSOpsVirtual);
-       localslot = table_slot_create(rel->localrel,
-                                                                 &estate->es_tupleTable);
-       EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1);
 
        /*
         * Populate updatedCols so that per-column triggers can fire.  This could
@@ -741,7 +756,6 @@ apply_handle_update(StringInfo s)
        fill_extraUpdatedCols(target_rte, RelationGetDescr(rel->localrel));
 
        PushActiveSnapshot(GetTransactionSnapshot());
-       ExecOpenIndices(estate->es_result_relation_info, false);
 
        /* Build the search tuple. */
        oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
@@ -749,20 +763,57 @@ apply_handle_update(StringInfo s)
                                                has_oldtup ? oldtup.values : newtup.values);
        MemoryContextSwitchTo(oldctx);
 
+       Assert(rel->localrel->rd_rel->relkind == RELKIND_RELATION);
+       apply_handle_update_internal(estate->es_result_relation_info, estate,
+                                                                remoteslot, &newtup, rel);
+
+       PopActiveSnapshot();
+
+       /* Handle queued AFTER triggers. */
+       AfterTriggerEndQuery(estate);
+
+       ExecResetTupleTable(estate->es_tupleTable, false);
+       FreeExecutorState(estate);
+
+       logicalrep_rel_close(rel, NoLock);
+
+       CommandCounterIncrement();
+}
+
+/* Workhorse for apply_handle_update() */
+static void
+apply_handle_update_internal(ResultRelInfo *relinfo,
+                                                        EState *estate, TupleTableSlot *remoteslot,
+                                                        LogicalRepTupleData *newtup,
+                                                        LogicalRepRelMapEntry *relmapentry)
+{
+       Relation        localrel = relinfo->ri_RelationDesc;
+       LogicalRepRelation *remoterel = &relmapentry->remoterel;
+       Oid                     idxoid;
+       EPQState        epqstate;
+       TupleTableSlot *localslot;
+       bool            found;
+       MemoryContext oldctx;
+
+       localslot = table_slot_create(localrel, &estate->es_tupleTable);
+       EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1);
+
+       ExecOpenIndices(relinfo, false);
+
        /*
         * Try to find tuple using either replica identity index, primary key or
         * if needed, sequential scan.
         */
-       idxoid = GetRelationIdentityOrPK(rel->localrel);
+       idxoid = GetRelationIdentityOrPK(localrel);
        Assert(OidIsValid(idxoid) ||
-                  (rel->remoterel.replident == REPLICA_IDENTITY_FULL && has_oldtup));
+                  (remoterel->replident == REPLICA_IDENTITY_FULL));
 
        if (OidIsValid(idxoid))
-               found = RelationFindReplTupleByIndex(rel->localrel, idxoid,
+               found = RelationFindReplTupleByIndex(localrel, idxoid,
                                                                                         LockTupleExclusive,
                                                                                         remoteslot, localslot);
        else
-               found = RelationFindReplTupleSeq(rel->localrel, LockTupleExclusive,
+               found = RelationFindReplTupleSeq(localrel, LockTupleExclusive,
                                                                                 remoteslot, localslot);
 
        ExecClearTuple(remoteslot);
@@ -776,8 +827,8 @@ apply_handle_update(StringInfo s)
        {
                /* Process and store remote tuple in the slot */
                oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
-               slot_modify_cstrings(remoteslot, localslot, rel,
-                                                        newtup.values, newtup.changed);
+               slot_modify_cstrings(remoteslot, localslot, relmapentry,
+                                                        newtup->values, newtup->changed);
                MemoryContextSwitchTo(oldctx);
 
                EvalPlanQualSetSlot(&epqstate, remoteslot);
@@ -795,23 +846,12 @@ apply_handle_update(StringInfo s)
                elog(DEBUG1,
                         "logical replication did not find row for update "
                         "in replication target relation \"%s\"",
-                        RelationGetRelationName(rel->localrel));
+                        RelationGetRelationName(localrel));
        }
 
        /* Cleanup. */
-       ExecCloseIndices(estate->es_result_relation_info);
-       PopActiveSnapshot();
-
-       /* Handle queued AFTER triggers. */
-       AfterTriggerEndQuery(estate);
-
+       ExecCloseIndices(relinfo);
        EvalPlanQualEnd(&epqstate);
-       ExecResetTupleTable(estate->es_tupleTable, false);
-       FreeExecutorState(estate);
-
-       logicalrep_rel_close(rel, NoLock);
-
-       CommandCounterIncrement();
 }
 
 /*
@@ -825,12 +865,8 @@ apply_handle_delete(StringInfo s)
        LogicalRepRelMapEntry *rel;
        LogicalRepTupleData oldtup;
        LogicalRepRelId relid;
-       Oid                     idxoid;
        EState     *estate;
-       EPQState        epqstate;
        TupleTableSlot *remoteslot;
-       TupleTableSlot *localslot;
-       bool            found;
        MemoryContext oldctx;
 
        ensure_transaction();
@@ -855,33 +891,64 @@ apply_handle_delete(StringInfo s)
        remoteslot = ExecInitExtraTupleSlot(estate,
                                                                                RelationGetDescr(rel->localrel),
                                                                                &TTSOpsVirtual);
-       localslot = table_slot_create(rel->localrel,
-                                                                 &estate->es_tupleTable);
-       EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1);
 
        PushActiveSnapshot(GetTransactionSnapshot());
-       ExecOpenIndices(estate->es_result_relation_info, false);
 
-       /* Find the tuple using the replica identity index. */
+       /* Build the search tuple. */
        oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
        slot_store_cstrings(remoteslot, rel, oldtup.values);
        MemoryContextSwitchTo(oldctx);
 
+       Assert(rel->localrel->rd_rel->relkind == RELKIND_RELATION);
+       apply_handle_delete_internal(estate->es_result_relation_info, estate,
+                                                                remoteslot, &rel->remoterel);
+
+       PopActiveSnapshot();
+
+       /* Handle queued AFTER triggers. */
+       AfterTriggerEndQuery(estate);
+
+       ExecResetTupleTable(estate->es_tupleTable, false);
+       FreeExecutorState(estate);
+
+       logicalrep_rel_close(rel, NoLock);
+
+       CommandCounterIncrement();
+}
+
+/* Workhorse for apply_handle_delete() */
+static void
+apply_handle_delete_internal(ResultRelInfo *relinfo, EState *estate,
+                                                        TupleTableSlot *remoteslot,
+                                                        LogicalRepRelation *remoterel)
+{
+       Relation        localrel = relinfo->ri_RelationDesc;
+       Oid                     idxoid;
+       EPQState        epqstate;
+       TupleTableSlot *localslot;
+       bool            found;
+
+       localslot = table_slot_create(localrel, &estate->es_tupleTable);
+       EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1);
+
+       ExecOpenIndices(relinfo, false);
+
        /*
         * Try to find tuple using either replica identity index, primary key or
         * if needed, sequential scan.
         */
-       idxoid = GetRelationIdentityOrPK(rel->localrel);
+       idxoid = GetRelationIdentityOrPK(localrel);
        Assert(OidIsValid(idxoid) ||
-                  (rel->remoterel.replident == REPLICA_IDENTITY_FULL));
+                  (remoterel->replident == REPLICA_IDENTITY_FULL));
 
        if (OidIsValid(idxoid))
-               found = RelationFindReplTupleByIndex(rel->localrel, idxoid,
+               found = RelationFindReplTupleByIndex(localrel, idxoid,
                                                                                         LockTupleExclusive,
                                                                                         remoteslot, localslot);
        else
-               found = RelationFindReplTupleSeq(rel->localrel, LockTupleExclusive,
+               found = RelationFindReplTupleSeq(localrel, LockTupleExclusive,
                                                                                 remoteslot, localslot);
+
        /* If found delete it. */
        if (found)
        {
@@ -896,23 +963,12 @@ apply_handle_delete(StringInfo s)
                elog(DEBUG1,
                         "logical replication could not find row for delete "
                         "in replication target relation \"%s\"",
-                        RelationGetRelationName(rel->localrel));
+                        RelationGetRelationName(localrel));
        }
 
        /* Cleanup. */
-       ExecCloseIndices(estate->es_result_relation_info);
-       PopActiveSnapshot();
-
-       /* Handle queued AFTER triggers. */
-       AfterTriggerEndQuery(estate);
-
+       ExecCloseIndices(relinfo);
        EvalPlanQualEnd(&epqstate);
-       ExecResetTupleTable(estate->es_tupleTable, false);
-       FreeExecutorState(estate);
-
-       logicalrep_rel_close(rel, NoLock);
-
-       CommandCounterIncrement();
 }
 
 /*