Skip to content

Commit

Permalink
Allow logical replication to copy tables in binary format.
Browse files Browse the repository at this point in the history 
This patch allows copying tables in the binary format during table
synchronization when the binary option for a subscription is enabled.
Previously, tables are copied in text format even if the subscription is
created with the binary option enabled. Copying tables in binary format
may reduce the time spent depending on column types.

A binary copy for initial table synchronization is supported only when
both publisher and subscriber are v16 or later.

Author: Melih Mutlu
Reviewed-by: Peter Smith, Shi yu, Euler Taveira, Vignesh C,  Kuroda Hayato, Osumi Takamichi, Bharath Rupireddy, Hou Zhijie
Discussion: https://postgr.es/m/CAGPVpCQvAziCLknEnygY0v1-KBtg%2BOm-9JHJYZOnNPKFJPompw%40mail.gmail.com
  • Loading branch information
Amit Kapila committed Mar 23, 2023
1 parent 4f1882b commit ecb6965
Show file tree
Hide file tree
Showing 5 changed files with 216 additions and 18 deletions.
4 changes: 3 additions & 1 deletion doc/src/sgml/logical-replication.sgml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -251,7 +251,9 @@
column of type <type>bigint</type>. The target table can also have
additional columns not provided by the published table. Any such columns
will be filled with the default value as specified in the definition of the
target table.
target table. However, logical replication in binary format is more
restrictive. See the <link linkend="sql-createsubscription-binary"><literal>binary</literal>
option</link> of <command>CREATE SUBSCRIPTION</command> for details.
</para>

<sect2 id="logical-replication-subscription-slot">
Expand Down
5 changes: 5 additions & 0 deletions doc/src/sgml/ref/alter_subscription.sgml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -177,6 +177,11 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
how <literal>copy_data = true</literal> can interact with the
<literal>origin</literal> parameter.
</para>
<para>
See the <link linkend="sql-createsubscription-binary"><literal>binary</literal>
option</link> of <command>CREATE SUBSCRIPTION</command> for details
about copying pre-existing data in binary format.
</para>
</listitem>
</varlistentry>
</variablelist></para>
Expand Down
28 changes: 22 additions & 6 deletions doc/src/sgml/ref/create_subscription.sgml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -185,15 +185,25 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl

<variablelist>

<varlistentry>
<varlistentry id="sql-createsubscription-binary" xreflabel="binary">
<term><literal>binary</literal> (<type>boolean</type>)</term>
<listitem>
<para>
Specifies whether the subscription will request the publisher to
send the data in binary format (as opposed to text).
The default is <literal>false</literal>.
Even when this option is enabled, only data types having
binary send and receive functions will be transferred in binary.
Specifies whether the subscription will request the publisher to send
the data in binary format (as opposed to text). The default is
<literal>false</literal>. Any initial table synchronization copy
(see <literal>copy_data</literal>) also uses the same format. Binary
format can be faster than the text format, but it is less portable
across machine architectures and <productname>PostgreSQL</productname>
versions. Binary format is very data type specific; for example, it
will not allow copying from a <type>smallint</type> column to an
<type>integer</type> column, even though that would work fine in text
format. Even when this option is enabled, only data types having binary
send and receive functions will be transferred in binary. Note that
the initial synchronization requires all data types to have binary
send and receive functions, otherwise the synchronization will fail
(see <xref linkend="sql-createtype"/> for more about send/receive
functions).
</para>

<para>
Expand All @@ -203,6 +213,12 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
such a case, data transfer will fail, and
the <literal>binary</literal> option cannot be used.
</para>

<para>
If the publisher is a <productname>PostgreSQL</productname> version
before 16, then any initial table synchronization will use text format
even if <literal>binary = true</literal>.
</para>
</listitem>
</varlistentry>

Expand Down
17 changes: 16 additions & 1 deletion src/backend/replication/logical/tablesync.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,7 @@
#include "catalog/pg_type.h"
#include "commands/copy.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "parser/parse_relation.h"
#include "pgstat.h"
#include "replication/logicallauncher.h"
Expand Down Expand Up @@ -1090,6 +1091,7 @@ copy_table(Relation rel)
CopyFromState cstate;
List *attnamelist;
ParseState *pstate;
List *options = NIL;

/* Get the publisher relation info. */
fetch_remote_table_info(get_namespace_name(RelationGetNamespace(rel)),
Expand Down Expand Up @@ -1168,6 +1170,19 @@ copy_table(Relation rel)

appendStringInfoString(&cmd, ") TO STDOUT");
}

/*
* Prior to v16, initial table synchronization will use text format even
* if the binary option is enabled for a subscription.
*/
if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 160000 &&
MySubscription->binary)
{
appendStringInfoString(&cmd, " WITH (FORMAT binary)");
options = list_make1(makeDefElem("format",
(Node *) makeString("binary"), -1));
}

res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 0, NULL);
pfree(cmd.data);
if (res->status != WALRCV_OK_COPY_OUT)
Expand All @@ -1184,7 +1199,7 @@ copy_table(Relation rel)
NULL, false, false);

attnamelist = make_copy_attnamelist(relmapentry);
cstate = BeginCopyFrom(pstate, rel, NULL, NULL, false, copy_read_data, attnamelist, NIL);
cstate = BeginCopyFrom(pstate, rel, NULL, NULL, false, copy_read_data, attnamelist, options);

/* Do the copy */
(void) CopyFrom(cstate);
Expand Down
180 changes: 170 additions & 10 deletions src/test/subscription/t/014_binary.pl
View file
Original file line number Diff line number Diff line change
Expand Up @@ -40,35 +40,72 @@
$node_publisher->safe_psql('postgres',
"CREATE PUBLICATION tpub FOR ALL TABLES");

# ------------------------------------------------------
# Ensure binary mode also executes COPY in binary format
# ------------------------------------------------------

# Insert some content before creating a subscription
$node_publisher->safe_psql(
'postgres', qq(
INSERT INTO public.test_numerical (a, b, c, d) VALUES
(1, 1.2, 1.3, 10),
(2, 2.2, 2.3, 20);
INSERT INTO public.test_arrays (a, b, c) VALUES
('{1,2,3}', '{1.1, 1.2, 1.3}', '{"one", "two", "three"}'),
('{3,1,2}', '{1.3, 1.1, 1.2}', '{"three", "one", "two"}');
));

my $publisher_connstring = $node_publisher->connstr . ' dbname=postgres';
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tsub CONNECTION '$publisher_connstring' "
. "PUBLICATION tpub WITH (slot_name = tpub_slot, binary = true)");

# Ensure the COPY command is executed in binary format on the publisher
$node_publisher->wait_for_log(
qr/LOG: ( [A-Z0-9]+:)? statement: COPY (.+)? TO STDOUT WITH \(FORMAT binary\)/
);

# Ensure nodes are in sync with each other
$node_subscriber->wait_for_subscription_sync($node_publisher, 'tsub');

my $sync_check = qq(
SELECT a, b, c, d FROM test_numerical ORDER BY a;
SELECT a, b, c FROM test_arrays ORDER BY a;
);

# Check the synced data on the subscriber
my $result = $node_subscriber->safe_psql('postgres', $sync_check);

is( $result, '1|1.2|1.3|10
2|2.2|2.3|20
{1,2,3}|{1.1,1.2,1.3}|{one,two,three}
{3,1,2}|{1.3,1.1,1.2}|{three,one,two}', 'check synced data on subscriber');

# ----------------------------------
# Ensure apply works in binary mode
# ----------------------------------

# Insert some content and make sure it's replicated across
$node_publisher->safe_psql(
'postgres', qq(
INSERT INTO public.test_arrays (a, b, c) VALUES
('{1,2,3}', '{1.1, 1.2, 1.3}', '{"one", "two", "three"}'),
('{3,1,2}', '{1.3, 1.1, 1.2}', '{"three", "one", "two"}');
('{2,1,3}', '{1.2, 1.1, 1.3}', '{"two", "one", "three"}'),
('{1,3,2}', '{1.1, 1.3, 1.2}', '{"one", "three", "two"}');
INSERT INTO public.test_numerical (a, b, c, d) VALUES
(1, 1.2, 1.3, 10),
(2, 2.2, 2.3, 20),
(3, 3.2, 3.3, 30);
(3, 3.2, 3.3, 30),
(4, 4.2, 4.3, 40);
));

$node_publisher->wait_for_catchup('tsub');

my $result = $node_subscriber->safe_psql('postgres',
$result = $node_subscriber->safe_psql('postgres',
"SELECT a, b, c, d FROM test_numerical ORDER BY a");

is( $result, '1|1.2|1.3|10
2|2.2|2.3|20
3|3.2|3.3|30', 'check replicated data on subscriber');
3|3.2|3.3|30
4|4.2|4.3|40', 'check replicated data on subscriber');

# Test updates as well
$node_publisher->safe_psql(
Expand All @@ -83,14 +120,21 @@
"SELECT a, b, c FROM test_arrays ORDER BY a");

is( $result, '{1,2,3}|{42,1.2,1.3}|
{1,3,2}|{42,1.3,1.2}|
{2,1,3}|{42,1.1,1.3}|
{3,1,2}|{42,1.1,1.2}|', 'check updated replicated data on subscriber');

$result = $node_subscriber->safe_psql('postgres',
"SELECT a, b, c, d FROM test_numerical ORDER BY a");

is( $result, '1|42||10
2|42||20
3|42||30', 'check updated replicated data on subscriber');
3|42||30
4|42||40', 'check updated replicated data on subscriber');

# ------------------------------------------------------------------------------
# Use ALTER SUBSCRIPTION to change to text format and then back to binary format
# ------------------------------------------------------------------------------

# Test to reset back to text formatting, and then to binary again
$node_subscriber->safe_psql('postgres',
Expand All @@ -99,7 +143,7 @@
$node_publisher->safe_psql(
'postgres', qq(
INSERT INTO public.test_numerical (a, b, c, d) VALUES
(4, 4.2, 4.3, 40);
(5, 5.2, 5.3, 50);
));

$node_publisher->wait_for_catchup('tsub');
Expand All @@ -110,7 +154,8 @@
is( $result, '1|42||10
2|42||20
3|42||30
4|4.2|4.3|40', 'check replicated data on subscriber');
4|42||40
5|5.2|5.3|50', 'check replicated data on subscriber');

$node_subscriber->safe_psql('postgres',
"ALTER SUBSCRIPTION tsub SET (binary = true);");
Expand All @@ -127,9 +172,124 @@
"SELECT a, b, c FROM test_arrays ORDER BY a");

is( $result, '{1,2,3}|{42,1.2,1.3}|
{1,3,2}|{42,1.3,1.2}|
{2,1,3}|{42,1.1,1.3}|
{2,3,1}|{1.2,1.3,1.1}|{two,three,one}
{3,1,2}|{42,1.1,1.2}|', 'check replicated data on subscriber');

# ---------------------------------------------------------------
# Test binary replication without and with send/receive functions
# ---------------------------------------------------------------

# Create a custom type without send/rcv functions
$ddl = qq(
CREATE TYPE myvarchar;
CREATE FUNCTION myvarcharin(cstring, oid, integer) RETURNS myvarchar
LANGUAGE internal IMMUTABLE PARALLEL SAFE STRICT AS 'varcharin';
CREATE FUNCTION myvarcharout(myvarchar) RETURNS cstring
LANGUAGE internal IMMUTABLE PARALLEL SAFE STRICT AS 'varcharout';
CREATE TYPE myvarchar (
input = myvarcharin,
output = myvarcharout);
CREATE TABLE public.test_myvarchar (
a myvarchar
););

$node_publisher->safe_psql('postgres', $ddl);
$node_subscriber->safe_psql('postgres', $ddl);

# Insert some initial data
$node_publisher->safe_psql(
'postgres', qq(
INSERT INTO public.test_myvarchar (a) VALUES
('a');
));

# Check the subscriber log from now on.
my $offset = -s $node_subscriber->logfile;

# Refresh the publication to trigger the tablesync
$node_subscriber->safe_psql('postgres',
"ALTER SUBSCRIPTION tsub REFRESH PUBLICATION");

# It should fail
$node_subscriber->wait_for_log(
qr/ERROR: ( [A-Z0-9]+:)? no binary input function available for type/,
$offset);

# Create and set send/rcv functions for the custom type
$ddl = qq(
CREATE FUNCTION myvarcharsend(myvarchar) RETURNS bytea
LANGUAGE internal STABLE PARALLEL SAFE STRICT AS 'varcharsend';
CREATE FUNCTION myvarcharrecv(internal, oid, integer) RETURNS myvarchar
LANGUAGE internal STABLE PARALLEL SAFE STRICT AS 'varcharrecv';
ALTER TYPE myvarchar SET (
send = myvarcharsend,
receive = myvarcharrecv
););

$node_publisher->safe_psql('postgres', $ddl);
$node_subscriber->safe_psql('postgres', $ddl);

# Now tablesync should succeed
$node_subscriber->wait_for_subscription_sync($node_publisher, 'tsub');

# Check the synced data on the subscriber
$result =
$node_subscriber->safe_psql('postgres', 'SELECT a FROM test_myvarchar;');

is($result, 'a', 'check synced data on subscriber with custom type');

# -----------------------------------------------------
# Test mismatched column types with/without binary mode
# -----------------------------------------------------

# Test syncing tables with mismatching column types
$node_publisher->safe_psql(
'postgres', qq(
CREATE TABLE public.test_mismatching_types (
a bigint PRIMARY KEY
);
INSERT INTO public.test_mismatching_types (a)
VALUES (1), (2);
));

# Check the subscriber log from now on.
$offset = -s $node_subscriber->logfile;

$node_subscriber->safe_psql(
'postgres', qq(
CREATE TABLE public.test_mismatching_types (
a int PRIMARY KEY
);
ALTER SUBSCRIPTION tsub REFRESH PUBLICATION;
));

# Cannot sync due to type mismatch
$node_subscriber->wait_for_log(
qr/ERROR: ( [A-Z0-9]+:)? incorrect binary data format/, $offset);

# Check the publisher log from now on.
$offset = -s $node_publisher->logfile;

# Setting binary to false should allow syncing
$node_subscriber->safe_psql(
'postgres', qq(
ALTER SUBSCRIPTION tsub SET (binary = false);));

# Ensure the COPY command is executed in text format on the publisher
$node_publisher->wait_for_log(
qr/LOG: ( [A-Z0-9]+:)? statement: COPY (.+)? TO STDOUT\n/, $offset);

$node_subscriber->wait_for_subscription_sync($node_publisher, 'tsub');

# Check the synced data on the subscriber
$result = $node_subscriber->safe_psql('postgres',
'SELECT a FROM test_mismatching_types ORDER BY a;');

is( $result, '1
2', 'check synced data on subscriber with binary = false');

$node_subscriber->stop('fast');
$node_publisher->stop('fast');

Expand Down

0 comments on commit ecb6965

Please sign in to comment.