Fix logical replication between different encodings

When sending a tuple attribute, the previous coding erroneously sent the
length byte before encoding conversion, which would lead to protocol
failures on the receiving side if the length did not match the following
string.

To fix that, use pq_sendcountedtext() for sending tuple attributes,
which takes care of all of that internally.  To match the API of
pq_sendcountedtext(), send even text values without a trailing zero byte
and have the receiving end put it in place instead.  This matches how
the standard FE/BE protocol behaves.

Reported-by: Kyotaro HORIGUCHI <horiguchi.kyotaro@lab.ntt.co.jp>

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index 5f971412ae70acc5ad3b6866f3f0dc6548f8f7f7..9d46d7411331c5703b378b8a95adea9e6e1ed63f 100644 (file)
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -6107,11 +6107,14 @@ TupleData
 </varlistentry>
 <varlistentry>
 <term>
-        String
+        Byte<replaceable>n</replaceable>
 </term>
 <listitem>
 <para>
-                The text value.
+                The value of the column, in text format.  (A future release
+                might support additional formats.)
+                <replaceable>n</replaceable> is the above length.
+
 </para>
 </listitem>
 </varlistentry>

diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index bc6e9b5a98cb726cd3697825f0d76e6a4abba192..bb607b6147a32b1e343d475c2b0725dade99cc59 100644 (file)
--- 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,10 +441,7 @@ 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 */
-
+               pq_sendcountedtext(out, outputstr, strlen(outputstr), false);
                pfree(outputstr);
 
                ReleaseSysCache(typtup);
@@ -492,7 +488,9 @@ logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple)
                                        len = pq_getmsgint(in, 4); /* read length */
 
                                        /* and data */
-                                       tuple->values[i] = (char *) pq_getmsgbytes(in, len);
+                                       tuple->values[i] = palloc(len + 1);
+                                       pq_copymsgbytes(in, tuple->values[i], len);
+                                       tuple->values[i][len] = '\0';
                                }
                                break;
                        default:

diff --git a/src/test/subscription/t/005_encoding.pl b/src/test/subscription/t/005_encoding.pl
new file mode 100644 (file)
index 0000000..42a4eee
--- /dev/null
+++ b/src/test/subscription/t/005_encoding.pl
@@ -0,0 +1,46 @@
+# Test replication between databases with different encodings
+use strict;
+use warnings;
+use PostgresNode;
+use TestLib;
+use Test::More tests => 1;
+
+sub wait_for_caught_up
+{
+       my ($node, $appname) = @_;
+
+       $node->poll_query_until('postgres',
+                                                       "SELECT pg_current_wal_location() <= replay_location FROM pg_stat_replication WHERE application_name = '$appname';")
+               or die "Timed out while waiting for subscriber to catch up";
+}
+
+my $node_publisher = get_new_node('publisher');
+$node_publisher->init(allows_streaming => 'logical', extra => ['--locale=C', '--encoding=UTF8']);
+$node_publisher->start;
+
+my $node_subscriber = get_new_node('subscriber');
+$node_subscriber->init(allows_streaming => 'logical', extra => ['--locale=C', '--encoding=LATIN1']);
+$node_subscriber->start;
+
+my $ddl = "CREATE TABLE test1 (a int, b text);";
+$node_publisher->safe_psql('postgres', $ddl);
+$node_subscriber->safe_psql('postgres', $ddl);
+
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+my $appname = 'encoding_test';
+
+$node_publisher->safe_psql('postgres', "CREATE PUBLICATION mypub FOR ALL TABLES;");
+$node_subscriber->safe_psql('postgres', "CREATE SUBSCRIPTION mysub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION mypub;");
+
+wait_for_caught_up($node_publisher, $appname);
+
+$node_publisher->safe_psql('postgres', q{INSERT INTO test1 VALUES (1, E'Mot\xc3\xb6rhead')});  # hand-rolled UTF-8
+
+wait_for_caught_up($node_publisher, $appname);
+
+is($node_subscriber->safe_psql('postgres', q{SELECT a FROM test1 WHERE b = E'Mot\xf6rhead'}),  # LATIN1
+                                                          qq(1),
+                                                          'data replicated to subscriber');
+
+$node_subscriber->stop;
+$node_publisher->stop;