diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c
index 088700e17cb910a12e5ddb2cbdbe8ea131f4866e..eef1dc2b1843de5376e415a12ec1eb86a6a56182 100644
--- a/src/backend/access/transam/parallel.c
+++ b/src/backend/access/transam/parallel.c
@@ -810,7 +810,17 @@ HandleParallelMessage(ParallelContext *pcxt, int i, StringInfo msg)
case 'A': /* NotifyResponse */
{
/* Propagate NotifyResponse. */
- pq_putmessage(msg->data[0], &msg->data[1], msg->len - 1);
+ int32 pid;
+ const char *channel;
+ const char *payload;
+
+ pid = pq_getmsgint(msg, 4);
+ channel = pq_getmsgrawstring(msg);
+ payload = pq_getmsgrawstring(msg);
+ pq_endmessage(msg);
+
+ NotifyMyFrontEnd(channel, payload, pid);
+
break;
}
@@ -988,6 +998,12 @@ ParallelWorkerMain(Datum main_arg)
BackgroundWorkerInitializeConnectionByOid(fps->database_id,
fps->authenticated_user_id);
+ /*
+ * Set the client encoding to the database encoding, since that is what
+ * the leader will expect.
+ */
+ SetClientEncoding(GetDatabaseEncoding());
+
/* Restore GUC values from launching backend. */
gucspace = shm_toc_lookup(toc, PARALLEL_KEY_GUC);
Assert(gucspace != NULL);
diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c
index c39ac3aeef00d506308034c85b43f8223177cdcf..716f1c33183da6e2cc0d50bf463098bfcbf213a6 100644
--- a/src/backend/commands/async.c
+++ b/src/backend/commands/async.c
@@ -390,9 +390,6 @@ static bool asyncQueueProcessPageEntries(volatile QueuePosition *current,
char *page_buffer);
static void asyncQueueAdvanceTail(void);
static void ProcessIncomingNotify(void);
-static void NotifyMyFrontEnd(const char *channel,
- const char *payload,
- int32 srcPid);
static bool AsyncExistsPendingNotify(const char *channel, const char *payload);
static void ClearPendingActionsAndNotifies(void);
@@ -2076,7 +2073,7 @@ ProcessIncomingNotify(void)
/*
* Send NOTIFY message to my front end.
*/
-static void
+void
NotifyMyFrontEnd(const char *channel, const char *payload, int32 srcPid)
{
if (whereToSendOutput == DestRemote)
diff --git a/src/backend/commands/variable.c b/src/backend/commands/variable.c
index 962d75db6e45be3afdd3059c6249bbc822eaea70..4ad8266a51c576edb7c60cf74b6c8cf14008caf3 100644
--- a/src/backend/commands/variable.c
+++ b/src/backend/commands/variable.c
@@ -755,6 +755,30 @@ assign_client_encoding(const char *newval, void *extra)
{
int encoding = *((int *) extra);
+ /*
+ * Parallel workers send data to the leader, not the client. They always
+ * send data using the database encoding.
+ */
+ if (IsParallelWorker())
+ {
+ /*
+ * During parallel worker startup, we want to accept the leader's
+ * client_encoding setting so that anyone who looks at the value in
+ * the worker sees the same value that they would see in the leader.
+ */
+ if (InitializingParallelWorker)
+ return;
+
+ /*
+ * A change other than during startup, for example due to a SET clause
+ * attached to a function definition, should be rejected, as there is
+ * nothing we can do inside the worker to make it take effect.
+ */
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_TRANSACTION_STATE),
+ errmsg("cannot change client_encoding in a parallel worker")));
+ }
+
/* We do not expect an error if PrepareClientEncoding succeeded */
if (SetClientEncoding(encoding) < 0)
elog(LOG, "SetClientEncoding(%d) failed", encoding);
diff --git a/src/backend/libpq/pqformat.c b/src/backend/libpq/pqformat.c
index 4ddea8285fc1758e330dd15d4def908004040722..b5d9d64e5479355759862df21db0787196831e45 100644
--- a/src/backend/libpq/pqformat.c
+++ b/src/backend/libpq/pqformat.c
@@ -65,6 +65,7 @@
* pq_copymsgbytes - copy raw data from a message buffer
* pq_getmsgtext - get a counted text string (with conversion)
* pq_getmsgstring - get a null-terminated text string (with conversion)
+ * pq_getmsgrawstring - get a null-terminated text string - NO conversion
* pq_getmsgend - verify message fully consumed
*/
@@ -639,6 +640,35 @@ pq_getmsgstring(StringInfo msg)
return pg_client_to_server(str, slen);
}
+/* --------------------------------
+ * pq_getmsgrawstring - get a null-terminated text string - NO conversion
+ *
+ * Returns a pointer directly into the message buffer.
+ * --------------------------------
+ */
+const char *
+pq_getmsgrawstring(StringInfo msg)
+{
+ char *str;
+ int slen;
+
+ str = &msg->data[msg->cursor];
+
+ /*
+ * It's safe to use strlen() here because a StringInfo is guaranteed to
+ * have a trailing null byte. But check we found a null inside the
+ * message.
+ */
+ slen = strlen(str);
+ if (msg->cursor + slen >= msg->len)
+ ereport(ERROR,
+ (errcode(ERRCODE_PROTOCOL_VIOLATION),
+ errmsg("invalid string in message")));
+ msg->cursor += slen + 1;
+
+ return str;
+}
+
/* --------------------------------
* pq_getmsgend - verify message fully consumed
* --------------------------------
diff --git a/src/backend/libpq/pqmq.c b/src/backend/libpq/pqmq.c
index 3225c1fa0e7f5c1673d13506710504a31e0ef7c3..0dcdee03db53fdd8584e5b4594afac5746b46c30 100644
--- a/src/backend/libpq/pqmq.c
+++ b/src/backend/libpq/pqmq.c
@@ -232,7 +232,7 @@ pq_parse_errornotice(StringInfo msg, ErrorData *edata)
pq_getmsgend(msg);
break;
}
- value = pq_getmsgstring(msg);
+ value = pq_getmsgrawstring(msg);
switch (code)
{
diff --git a/src/include/commands/async.h b/src/include/commands/async.h
index b4c13fac4a311498c020293b27058a5b90909d35..95559df19fea933ce1fb85da4ec166430062963e 100644
--- a/src/include/commands/async.h
+++ b/src/include/commands/async.h
@@ -28,6 +28,10 @@ extern volatile sig_atomic_t notifyInterruptPending;
extern Size AsyncShmemSize(void);
extern void AsyncShmemInit(void);
+extern void NotifyMyFrontEnd(const char *channel,
+ const char *payload,
+ int32 srcPid);
+
/* notify-related SQL statements */
extern void Async_Notify(const char *channel, const char *payload);
extern void Async_Listen(const char *channel);
diff --git a/src/include/libpq/pqformat.h b/src/include/libpq/pqformat.h
index 65ebf37fbc0bae12e592f01bc2f7ddb7fe05a34a..3c0d4b2622bfffe9f2745871d98db10b9daef2b0 100644
--- a/src/include/libpq/pqformat.h
+++ b/src/include/libpq/pqformat.h
@@ -44,6 +44,7 @@ extern const char *pq_getmsgbytes(StringInfo msg, int datalen);
extern void pq_copymsgbytes(StringInfo msg, char *buf, int datalen);
extern char *pq_getmsgtext(StringInfo msg, int rawbytes, int *nbytes);
extern const char *pq_getmsgstring(StringInfo msg);
+extern const char *pq_getmsgrawstring(StringInfo msg);
extern void pq_getmsgend(StringInfo msg);
#endif /* PQFORMAT_H */