0001-Ensure-that-pg_create_physical_replication_slot-erro.patch

From 290f612ab41f6cb779d44c5754b23092d139e88d Mon Sep 17 00:00:00 2001
From: Dave Cramer <davecramer@gmail.com>
Date: Tue, 31 Jul 2018 14:29:09 -0400
Subject: [PATCH 1/2] Ensure that pg_create_physical_replication_slot errors if
 the output plugin does not exist If the output plugin does not exist send
 back an error message as opposed to CopyBoth when the starting the
 replication stream

---
 contrib/test_decoding/expected/slot.out   |  2 ++
 contrib/test_decoding/sql/slot.sql        |  2 ++
 src/backend/replication/logical/logical.c |  5 ++---
 src/backend/replication/walsender.c       | 22 ++++++++++++----------
 4 files changed, 18 insertions(+), 13 deletions(-)

diff --git a/contrib/test_decoding/expected/slot.out b/contrib/test_decoding/expected/slot.out
index 2737a8a..523621a 100644
--- a/contrib/test_decoding/expected/slot.out
+++ b/contrib/test_decoding/expected/slot.out
@@ -30,6 +30,8 @@ SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot_t2', 'tes
  init
 (1 row)
 
+SELECT pg_create_logical_replication_slot('foo', 'nonexistent');
+ERROR:  could not access file "nonexistent": No such file or directory
 -- here we want to start a new session and wait till old one is gone
 select pg_backend_pid() as oldpid \gset
 \c -
diff --git a/contrib/test_decoding/sql/slot.sql b/contrib/test_decoding/sql/slot.sql
index 24cdf71..c8d08f8 100644
--- a/contrib/test_decoding/sql/slot.sql
+++ b/contrib/test_decoding/sql/slot.sql
@@ -9,6 +9,8 @@ SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot_p', 'test
 
 SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot_t2', 'test_decoding', true);
 
+SELECT pg_create_logical_replication_slot('foo', 'nonexistent');
+
 -- here we want to start a new session and wait till old one is gone
 select pg_backend_pid() as oldpid \gset
 \c -
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 3cd4eef..9f883b9 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -143,8 +143,7 @@ StartupDecodingContext(List *output_plugin_options,
 	 * (re-)load output plugins, so we detect a bad (removed) output plugin
 	 * now.
 	 */
-	if (!fast_forward)
-		LoadOutputPlugin(&ctx->callbacks, NameStr(slot->data.plugin));
+	LoadOutputPlugin(&ctx->callbacks, NameStr(slot->data.plugin));
 
 	/*
 	 * Now that the slot's xmin has been set, we can announce ourselves as a
@@ -312,7 +311,7 @@ CreateInitDecodingContext(char *plugin,
 	ReplicationSlotSave();
 
 	ctx = StartupDecodingContext(NIL, InvalidXLogRecPtr, xmin_horizon,
-								 need_full_snapshot, true,
+								 need_full_snapshot, true, 
 								 read_page, prepare_write, do_write,
 								 update_progress);
 
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index d60026d..a196c5b 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1068,6 +1068,18 @@ StartLogicalReplication(StartReplicationCmd *cmd)
 		got_STOPPING = true;
 	}
 
+	/*
+	* Initialize position to the last ack'ed one, then the xlog records begin
+	* to be shipped from that position.
+	*/
+	logical_decoding_ctx = CreateDecodingContext(cmd->startpoint, cmd->options,
+											 false,
+											 logical_read_xlog_page,
+											 WalSndPrepareWrite,
+											 WalSndWriteData,
+											 WalSndUpdateProgress);
+
+
 	WalSndSetState(WALSNDSTATE_CATCHUP);
 
 	/* Send a CopyBothResponse message, and start streaming */
@@ -1077,16 +1089,6 @@ StartLogicalReplication(StartReplicationCmd *cmd)
 	pq_endmessage(&buf);
 	pq_flush();
 
-	/*
-	 * Initialize position to the last ack'ed one, then the xlog records begin
-	 * to be shipped from that position.
-	 */
-	logical_decoding_ctx = CreateDecodingContext(cmd->startpoint, cmd->options,
-												 false,
-												 logical_read_xlog_page,
-												 WalSndPrepareWrite,
-												 WalSndWriteData,
-												 WalSndUpdateProgress);
 
 	/* Start reading WAL from the oldest required WAL. */
 	logical_startptr = MyReplicationSlot->data.restart_lsn;
-- 
2.6.4