changes_amit_v53_1.patch

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 674ff9b9a3..df3aad43e1 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -863,16 +863,16 @@ HandleParallelApplyMessages(void)
 		void	   *data;
 		ParallelApplyWorkerInfo *winfo = (ParallelApplyWorkerInfo *) lfirst(lc);
 
-		res = shm_mq_receive(winfo->error_mq_handle, &nbytes, &data, true);
-
 		/*
 		 * The leader will detach from the error queue and set it to NULL
 		 * before preparing to stop all parallel apply workers, so we don't
-		 * need to handle error messages anymore.
+		 * need to handle error messages anymore. See logicalrep_worker_detach.
 		 */
 		if (!winfo->error_mq_handle)
 			continue;
 
+		res = shm_mq_receive(winfo->error_mq_handle, &nbytes, &data, true);
+
 		if (res == SHM_MQ_WOULD_BLOCK)
 			continue;
 		else if (res == SHM_MQ_SUCCESS)
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 95036c5f2b..1368cf2dde 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -696,8 +696,8 @@ logicalrep_worker_detach(void)
 		/*
 		 * Detach from the error_mq_handle for all parallel apply workers
 		 * before terminating them to prevent the leader apply worker from
-		 * receiving the worker termination messages which will cause the
-		 * leader to exit.
+		 * receiving the worker termination messages and sending it to logs
+		 * when the same is already done by individual parallel worker.
 		 */
 		pa_detach_all_error_mq();
 
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 35deb3126c..56e3c6c1a2 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -561,19 +561,20 @@ end_replication_step(void)
 }
 
 /*
- * Handle streamed transactions for both the leader apply worker and the parallel
- * apply workers.
+ * Handle streamed transactions for both the leader apply worker and the
+ * parallel apply workers.
  *
- * In streaming case (receiving a block of streamed transaction), for serialize
- * mode, simply redirect it to a file for the proper toplevel transaction, and
- * for parallel mode, the leader apply worker will send the changes to parallel
- * apply workers and the parallel apply worker will define savepoints if
- * needed. (LOGICAL_REP_MSG_RELATION or LOGICAL_REP_MSG_TYPE changes will be
- * applied by both leader apply worker and parallel apply workers).
+ * In the streaming case (receiving a block of the streamed transaction), for
+ * serialize mode, simply redirect it to a file for the proper toplevel
+ * transaction, and for parallel mode, the leader apply worker will send the
+ * changes to parallel apply workers and the parallel apply worker will define
+ * savepoints if needed. (LOGICAL_REP_MSG_RELATION or LOGICAL_REP_MSG_TYPE
+ * changes will be applied by both leader apply worker and parallel apply
+ * workers).
  *
- * For non-streamed transactions, returns false.
- * For streamed transactions, returns true if in leader apply worker, false
- * otherwise.
+ * Returns true for streamed transactions (when the change is either serialized
+ * to file or sent to parallel apply worker), false otherwise (regular mode or
+ * needs to be processed by parallel apply worker).
  *
  * Exception: If the message being processed is LOGICAL_REP_MSG_RELATION
  * or LOGICAL_REP_MSG_TYPE, return false even if the message needs to be sent
@@ -1392,8 +1393,6 @@ apply_handle_origin(StringInfo s)
 				 errmsg_internal("ORIGIN message sent out of order")));
 }
 
-
-
 /*
  * Handle STREAM START message.
  */
@@ -1558,7 +1557,8 @@ apply_handle_stream_stop(StringInfo s)
 			/*
 			 * Lock before sending the STREAM_STOP message so that the leader
 			 * can hold the lock first and the parallel apply worker will wait
-			 * for leader to release the lock.
+			 * for leader to release the lock. See Locking Considerations atop
+			 * applyparallelworker.c.
 			 */
 			pa_lock_stream(winfo->shared->xid, AccessExclusiveLock);
 
@@ -1576,8 +1576,11 @@ apply_handle_stream_stop(StringInfo s)
 			/*
 			 * By the time parallel apply worker is processing the changes in
 			 * the current streaming block, the leader apply worker may have
-			 * sent multiple streaming blocks. So, try to lock only if there
-			 * is no message left in the queue.
+			 * sent multiple streaming blocks. This can lead to parallel apply
+			 * worker start waiting even when there are more chunk of streams
+			 * in the queue. So, try to lock only if there is no message left
+			 * in the queue. See Locking Considerations atop
+			 * applyparallelworker.c.
 			 */
 			if (pg_atomic_sub_fetch_u32(&MyParallelShared->pending_stream_count, 1) == 0)
 			{
@@ -1733,9 +1736,11 @@ apply_handle_stream_abort(StringInfo s)
 			 * XXX For the case of aborting the subtransaction, we only
 			 * increment the number of streaming blocks without releasing the
 			 * lock. This may slightly delay the processing of STREAM_ABORT
-			 * message but can ensure that the parallel apply worker will wait
-			 * on the lock for the next set of changes after processing the
-			 * STREAM_ABORT message.
+			 * message as the parallel apply worker may already be waiting on a
+			 * lock after processing STREAM_STOP message but can ensure that the
+			 * parallel apply worker will wait on the lock for the next set of
+			 * changes after processing the STREAM_ABORT message if it is not
+			 * already waiting for STREAM_STOP message.
 			 */
 			if (toplevel_xact)
 				pa_unlock_stream(xid, AccessExclusiveLock);
@@ -1759,8 +1764,11 @@ apply_handle_stream_abort(StringInfo s)
 			 *
 			 * By the time parallel apply worker is processing the changes in
 			 * the current streaming block, the leader apply worker may have
-			 * sent multiple streaming blocks. So, try to lock only if there is
-			 * no message left in the queue.
+			 * sent multiple streaming blocks. This can lead to parallel apply
+			 * worker start waiting even when there are more chunk of streams
+			 * in the queue. So, try to lock only if there is no message left
+			 * in the queue. See Locking Considerations atop
+			 * applyparallelworker.c.
 			 */
 			if (!toplevel_xact &&
 				pg_atomic_sub_fetch_u32(&MyParallelShared->pending_stream_count, 1) == 0)
@@ -3156,7 +3164,10 @@ store_flush_position(XLogRecPtr remote_lsn)
 {
 	FlushPosition *flushpos;
 
-	/* Skip for parallel apply workers. */
+	/*
+	 * Skip for parallel apply workers. The leader apply worker will ensure to
+	 * update it as the lsn_mapping is maintained by it.
+	 */
 	if (am_parallel_apply_worker())
 		return;