Re: [HACKERS] Logical Replication and Character encoding

At Fri, 3 Mar 2017 21:31:24 -0500, Peter Eisentraut  wrote in <88397afa-a8ec-8d8a-1c94-94a4795a3870@2ndquadrant.com>
> On 3/3/17 14:51, Petr Jelinek wrote:
> > On 03/03/17 20:37, Peter Eisentraut wrote:
> >> On 2/27/17 00:23, Kyotaro HORIGUCHI wrote:
> >>> Yeah, the patch sends converted string with the length of the
> >>> orignal length. Usually encoding conversion changes the length of
> >>> a string. I doubt that the reverse case was working correctly.
> >>
> >> I think we shouldn't send the length value at all.  This might have been
> >> a leftover from an earlier version of the patch.
> >>
> >> See attached patch that removes the length value.
> >>
> > 
> > Well the length is necessary to be able to add binary format support in
> > future so it's definitely not an omission.
> 
> Right.  So it appears the right function to use here is
> pq_sendcountedtext().  However, this sends the strings without null
> termination, so we'd have to do extra processing on the receiving end.
> Or we could add an option to pq_sendcountedtext() to make it do what we
> want.  I'd rather stick to standard pqformat.c functions for handling
> the protocol.

It seems reasonable. I changed the prototype of
pg_sendcountedtext as the following,

| extern void pq_sendcountedtext(StringInfo buf, const char *str, int slen,
| 			   bool countincludesself, bool binary);

I think 'binary' seems fine for the parameter here. The patch
size increased a bit.

By the way, I noticed that postmaster launches logical
replication launcher even if wal_level < logical. Is it right?

regards,

-- 
Kyotaro Horiguchi
NTT Open Source Software Center
diff --git a/src/backend/access/common/printsimple.c b/src/backend/access/common/printsimple.c
index 5fe1c72..62fa70d 100644
--- a/src/backend/access/common/printsimple.c
+++ b/src/backend/access/common/printsimple.c
@@ -96,7 +96,7 @@ printsimple(TupleTableSlot *slot, DestReceiver *self)					pq_sendcountedtext(&buf,									   VARDATA_ANY(t),									   VARSIZE_ANY_EXHDR(t),
-									   false);
+									   false, false);				}				break;
@@ -106,7 +106,7 @@ printsimple(TupleTableSlot *slot, DestReceiver *self)					char	str[12];	/* sign, 10 digits and '\0' */					pg_ltoa(num, str);
-					pq_sendcountedtext(&buf, str, strlen(str), false);
+					pq_sendcountedtext(&buf, str, strlen(str), false, false);				}				break;
@@ -116,7 +116,7 @@ printsimple(TupleTableSlot *slot, DestReceiver *self)					char	str[23];	/* sign, 21 digits and '\0' */					pg_lltoa(num, str);
-					pq_sendcountedtext(&buf, str, strlen(str), false);
+					pq_sendcountedtext(&buf, str, strlen(str), false, false);				}				break;
diff --git a/src/backend/access/common/printtup.c b/src/backend/access/common/printtup.c
index a2ca2d7..1ebc50f 100644
--- a/src/backend/access/common/printtup.c
+++ b/src/backend/access/common/printtup.c
@@ -353,7 +353,8 @@ printtup(TupleTableSlot *slot, DestReceiver *self)			char	   *outputstr;			outputstr = OutputFunctionCall(&thisState->finfo, attr);
-			pq_sendcountedtext(&buf, outputstr, strlen(outputstr), false);
+			pq_sendcountedtext(&buf, outputstr, strlen(outputstr),
+							   false, false);		}		else		{
@@ -442,7 +443,7 @@ printtup_20(TupleTableSlot *slot, DestReceiver *self)		Assert(thisState->format == 0);		outputstr = OutputFunctionCall(&thisState->finfo, attr);
-		pq_sendcountedtext(&buf, outputstr, strlen(outputstr), true);
+		pq_sendcountedtext(&buf, outputstr, strlen(outputstr), true, false);	}	pq_endmessage(&buf);
diff --git a/src/backend/libpq/pqformat.c b/src/backend/libpq/pqformat.c
index c8cf67c..f799130 100644
--- a/src/backend/libpq/pqformat.c
+++ b/src/backend/libpq/pqformat.c
@@ -129,23 +129,24 @@ pq_sendbytes(StringInfo buf, const char *data, int datalen) */voidpq_sendcountedtext(StringInfo buf, const char *str, int slen,
-				   bool countincludesself)
+				   bool countincludesself, bool binary){
-	int			extra = countincludesself ? 4 : 0;
+	int			extra_self		= countincludesself ? 4 : 0;
+	int			extra_binary	= binary ? 1 : 0;	char	   *p;	p = pg_server_to_client(str, slen);	if (p != str)				/* actual conversion has been done? */	{		slen = strlen(p);
-		pq_sendint(buf, slen + extra, 4);
-		appendBinaryStringInfo(buf, p, slen);
+		pq_sendint(buf, slen + extra_self + extra_binary, 4);
+		appendBinaryStringInfo(buf, p, slen + extra_binary);		pfree(p);	}	else	{
-		pq_sendint(buf, slen + extra, 4);
-		appendBinaryStringInfo(buf, str, slen);
+		pq_sendint(buf, slen + extra_self + extra_binary, 4);
+		appendBinaryStringInfo(buf, str, slen + extra_binary);	}}
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index bc6e9b5..c5e4214 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -417,7 +417,6 @@ logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple)		Form_pg_type typclass;		Form_pg_attribute att = desc->attrs[i];		char	   *outputstr;
-		int			len;		/* skip dropped columns */		if (att->attisdropped)
@@ -442,9 +441,12 @@ logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple)		pq_sendbyte(out, 't');	/* 'text' data follows */		outputstr = OidOutputFunctionCall(typclass->typoutput, values[i]);
-		len = strlen(outputstr) + 1;	/* null terminated */
-		pq_sendint(out, len, 4);		/* length */
-		pq_sendstring(out, outputstr);	/* data */
+
+		/*
+		 * Here, the string is sent as binary data so the length field counts
+		 * terminating NULL.
+		 */
+		pq_sendcountedtext(out, outputstr, strlen(outputstr), false, true);		pfree(outputstr);
diff --git a/src/backend/tcop/fastpath.c b/src/backend/tcop/fastpath.c
index 2efed95..0ca9522 100644
--- a/src/backend/tcop/fastpath.c
+++ b/src/backend/tcop/fastpath.c
@@ -160,7 +160,8 @@ SendFunctionResult(Datum retval, bool isnull, Oid rettype, int16 format)			getTypeOutputInfo(rettype, &typoutput, &typisvarlena);			outputstr = OidOutputFunctionCall(typoutput, retval);
-			pq_sendcountedtext(&buf, outputstr, strlen(outputstr), false);
+			pq_sendcountedtext(&buf, outputstr, strlen(outputstr),
+							   false, false);			pfree(outputstr);		}		else if (format == 1)
diff --git a/src/include/libpq/pqformat.h b/src/include/libpq/pqformat.h
index 4df87ec..29cba1a 100644
--- a/src/include/libpq/pqformat.h
+++ b/src/include/libpq/pqformat.h
@@ -19,7 +19,7 @@ extern void pq_beginmessage(StringInfo buf, char msgtype);extern void pq_sendbyte(StringInfo buf, int byt);extern void pq_sendbytes(StringInfo buf, const char *data, int datalen);extern void pq_sendcountedtext(StringInfo buf, const char *str, int slen,
-				   bool countincludesself);
+					   bool countincludesself, bool binary);extern void pq_sendtext(StringInfo buf, const char *str, int slen);extern void pq_sendstring(StringInfo buf, const char *str);extern void pq_send_ascii_string(StringInfo buf, const char *str);