Allow using replication origins in SQL level parallel sessions

diff --git a/doc/src/sgml/func/func-admin.sgml b/doc/src/sgml/func/func-admin.sgml
index 57ff333159f..64effa61d77 100644
--- a/doc/src/sgml/func/func-admin.sgml
+++ b/doc/src/sgml/func/func-admin.sgml
@@ -1308,21 +1308,21 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
         no such replication origin is found, <literal>NULL</literal> is
         returned.
        </para></entry>
       </row>
 
       <row>
        <entry id="pg-replication-origin-session-setup" role="func_table_entry"><para role="func_signature">
         <indexterm>
          <primary>pg_replication_origin_session_setup</primary>
         </indexterm>
-        <function>pg_replication_origin_session_setup</function> ( <parameter>node_name</parameter> <type>text</type> )
+        <function>pg_replication_origin_session_setup</function> ( <parameter>node_name</parameter> <type>text</type> <optional>, <parameter>acquired_by</parameter> <type>integer</type></optional> )
         <returnvalue>void</returnvalue>
        </para>
        <para>
         Marks the current session as replaying from the given
         origin, allowing replay progress to be tracked.
         Can only be used if no origin is currently selected.
         Use <function>pg_replication_origin_session_reset</function> to undo.
        </para></entry>
       </row>
 
diff --git a/src/backend/replication/logical/origin.c b/src/backend/replication/logical/origin.c
index 87f10e50dcc..7022422988a 100644
--- a/src/backend/replication/logical/origin.c
+++ b/src/backend/replication/logical/origin.c
@@ -1381,20 +1381,44 @@ pg_replication_origin_session_setup(PG_FUNCTION_ARGS)
 	origin = replorigin_by_name(name, false);
 	replorigin_session_setup(origin, 0);
 
 	replorigin_session_origin = origin;
 
 	pfree(name);
 
 	PG_RETURN_VOID();
 }
 
+/*
+ * Setup a replication origin for this session acquired by another
+ */
+Datum
+pg_replication_origin_session_setup_acquired_by(PG_FUNCTION_ARGS)
+{
+	char	   *name;
+	RepOriginId origin;
+	int			acquired_by;
+
+	replorigin_check_prerequisites(true, false);
+
+	name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
+	acquired_by = PG_GETARG_INT32(1);
+	origin = replorigin_by_name(name, false);
+	replorigin_session_setup(origin, acquired_by);
+
+	replorigin_session_origin = origin;
+
+	pfree(name);
+
+	PG_RETURN_VOID();
+}
+
 /*
  * Reset previously setup origin in this session
  */
 Datum
 pg_replication_origin_session_reset(PG_FUNCTION_ARGS)
 {
 	replorigin_check_prerequisites(true, false);
 
 	replorigin_session_reset();
 
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 118d6da1ace..7cd580164cb 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -12218,20 +12218,25 @@
   descr => 'translate the replication origin\'s name to its id',
   proname => 'pg_replication_origin_oid', provolatile => 's',
   prorettype => 'oid', proargtypes => 'text',
   prosrc => 'pg_replication_origin_oid' },
 
 { oid => '6006',
   descr => 'configure session to maintain replication progress tracking for the passed in origin',
   proname => 'pg_replication_origin_session_setup', provolatile => 'v',
   proparallel => 'u', prorettype => 'void', proargtypes => 'text',
   prosrc => 'pg_replication_origin_session_setup' },
+{ oid => '6015',
+  descr => 'configure session to maintain replication progress tracking for the passed in origin',
+  proname => 'pg_replication_origin_session_setup', provolatile => 'v',
+  proparallel => 'u', prorettype => 'void', proargtypes => 'text int4',
+  prosrc => 'pg_replication_origin_session_setup_acquired_by' },
 
 { oid => '6007', descr => 'teardown configured replication progress tracking',
   proname => 'pg_replication_origin_session_reset', provolatile => 'v',
   proparallel => 'u', prorettype => 'void', proargtypes => '',
   prosrc => 'pg_replication_origin_session_reset' },
 
 { oid => '6008',
   descr => 'is a replication origin configured in this session',
   proname => 'pg_replication_origin_session_is_setup', provolatile => 'v',
   proparallel => 'r', prorettype => 'bool', proargtypes => '',