Re: [Patch] add new parameter to pg_replication_origin_session_setup

[Amit Kapila]

On Thu, Sep 18, 2025 at 1:07 PM Hayato Kuroda (Fujitsu)
<kuroda.hay...@fujitsu.com> wrote:
>
> Dear hackers,
>
> > I considered a test, please see attached files.
>

Few comments:
1. +step "s0_compare" {
+    SELECT s0.lsn < s1.lsn
+    FROM local_lsn_store as s0, local_lsn_store as s1
+    WHERE s0.session = 0 AND s1.session = 1;
+}

This appears to be a bit tricky to compare the values. Doing a
sequential scan won't guarantee the order of rows' appearance. Can't
we somehow get the two rows ordered by session_id and compare their
values?

2.
+ else if (candidate_state->acquired_by != acquired_by)
+ {
+ if (initialized)
+ candidate_state->roident = InvalidRepOriginId;
+
  elog(ERROR, "could not find replication state slot for replication
origin with OID %u which was acquired by %d",
  node, acquired_by);
+ }

This doesn't appear neat. Instead, how about checking this case before
setting current_state as shown in attached. If we do that, we
shouldn't even need new variables like current_state and initialized.
Additionally, as shown in attached, it is better to make this a
user-facing error by using ereport.

3. Merge all patches as I don't see the need to do any backpatch here.

-- 
With Regards,
Amit Kapila.

diff --git a/src/backend/replication/logical/origin.c 
b/src/backend/replication/logical/origin.c
index 0bbc96bcee5..c93b6eb1798 100644
--- a/src/backend/replication/logical/origin.c
+++ b/src/backend/replication/logical/origin.c
@@ -1169,6 +1169,15 @@ replorigin_session_setup(RepOriginId node, int 
acquired_by)
                                                        curstate->roident, 
curstate->acquired_by)));
                }
 
+               else if (curstate->acquired_by != 0 &&
+                                curstate->acquired_by != acquired_by)
+               {
+                       ereport(ERROR,
+                                       
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+                                        errmsg("could not find replication 
state slot for replication origin with OID %u which was acquired by %d",
+                                                       node, acquired_by)));
+               }
+
                /* ok, found slot */
                candidate_state = curstate;
                break;
@@ -1196,14 +1205,8 @@ replorigin_session_setup(RepOriginId node, int 
acquired_by)
 
        if (acquired_by == 0)
                candidate_state->acquired_by = MyProcPid;
-       else if (candidate_state->acquired_by != acquired_by)
-       {
-               if (initialized)
-                       candidate_state->roident = InvalidRepOriginId;
-
-               elog(ERROR, "could not find replication state slot for 
replication origin with OID %u which was acquired by %d",
-                        node, acquired_by);
-       }
+       else
+               Assert(candidate_state->acquired_by == acquired_by);
 
        /* Candidate slot looks ok, use it */
        session_replication_state = candidate_state;