Refactor code to look up local replication tuple

This unifies some duplicate code.

Author: Amit Langote <amitlangote09@gmail.com>
Discussion: https://www.postgresql.org/message-id/CA+HiwqFjYE5anArxvkjr37AQMd52L-LZtz9Ld2QrLQ3YfcYhTw@mail.gmail.com

diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index fa3811783f6e9ecd7719a8de8c3813b3bb243315..673ebd211d1832c00a72423f9666acc850a651b7 100644 (file)
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -122,6 +122,10 @@ static void apply_handle_update_internal(ResultRelInfo *relinfo,
 static void apply_handle_delete_internal(ResultRelInfo *relinfo, EState *estate,
                                                                                 TupleTableSlot *remoteslot,
                                                                                 LogicalRepRelation *remoterel);
+static bool FindReplTupleInLocalRel(EState *estate, Relation localrel,
+                                                                       LogicalRepRelation *remoterel,
+                                                                       TupleTableSlot *remoteslot,
+                                                                       TupleTableSlot **localslot);
 
 /*
  * Should this worker apply changes for given relation.
@@ -788,33 +792,17 @@ apply_handle_update_internal(ResultRelInfo *relinfo,
                                                         LogicalRepRelMapEntry *relmapentry)
 {
        Relation        localrel = relinfo->ri_RelationDesc;
-       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(localrel);
-       Assert(OidIsValid(idxoid) ||
-                  (relmapentry->remoterel.replident == REPLICA_IDENTITY_FULL));
-
-       if (OidIsValid(idxoid))
-               found = RelationFindReplTupleByIndex(localrel, idxoid,
-                                                                                        LockTupleExclusive,
-                                                                                        remoteslot, localslot);
-       else
-               found = RelationFindReplTupleSeq(localrel, LockTupleExclusive,
-                                                                                remoteslot, localslot);
-
+       found = FindReplTupleInLocalRel(estate, localrel,
+                                                                       &relmapentry->remoterel,
+                                                                       remoteslot, &localslot);
        ExecClearTuple(remoteslot);
 
        /*
@@ -922,31 +910,15 @@ apply_handle_delete_internal(ResultRelInfo *relinfo, EState *estate,
                                                         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(localrel);
-       Assert(OidIsValid(idxoid) ||
-                  (remoterel->replident == REPLICA_IDENTITY_FULL));
-
-       if (OidIsValid(idxoid))
-               found = RelationFindReplTupleByIndex(localrel, idxoid,
-                                                                                        LockTupleExclusive,
-                                                                                        remoteslot, localslot);
-       else
-               found = RelationFindReplTupleSeq(localrel, LockTupleExclusive,
-                                                                                remoteslot, localslot);
+       found = FindReplTupleInLocalRel(estate, localrel, remoterel,
+                                                                       remoteslot, &localslot);
 
        /* If found delete it. */
        if (found)
@@ -970,6 +942,39 @@ apply_handle_delete_internal(ResultRelInfo *relinfo, EState *estate,
        EvalPlanQualEnd(&epqstate);
 }
 
+/*
+ * Try to find a tuple received from the publication side (in 'remoteslot') in
+ * the corresponding local relation using either replica identity index,
+ * primary key or if needed, sequential scan.
+ *
+ * Local tuple, if found, is returned in '*localslot'.
+ */
+static bool
+FindReplTupleInLocalRel(EState *estate, Relation localrel,
+                                               LogicalRepRelation *remoterel,
+                                               TupleTableSlot *remoteslot,
+                                               TupleTableSlot **localslot)
+{
+       Oid                     idxoid;
+       bool            found;
+
+       *localslot = table_slot_create(localrel, &estate->es_tupleTable);
+
+       idxoid = GetRelationIdentityOrPK(localrel);
+       Assert(OidIsValid(idxoid) ||
+                  (remoterel->replident == REPLICA_IDENTITY_FULL));
+
+       if (OidIsValid(idxoid))
+               found = RelationFindReplTupleByIndex(localrel, idxoid,
+                                                                                        LockTupleExclusive,
+                                                                                        remoteslot, *localslot);
+       else
+               found = RelationFindReplTupleSeq(localrel, LockTupleExclusive,
+                                                                                remoteslot, *localslot);
+
+       return found;
+}
+
 /*
  * Handle TRUNCATE message.
  *