public inbox for [email protected]
help / color / mirror / Atom feedRE: Allow logical replication to copy tables in binary format
14+ messages / 8 participants
[nested] [flat]
* RE: Allow logical replication to copy tables in binary format
@ 2023-02-20 07:12 Hayato Kuroda (Fujitsu) <[email protected]>
0 siblings, 1 reply; 14+ messages in thread
From: Hayato Kuroda (Fujitsu) @ 2023-02-20 07:12 UTC (permalink / raw)
To: 'Melih Mutlu' <[email protected]>; +Cc: Takamichi Osumi (Fujitsu) <[email protected]>; [email protected] <[email protected]>; Euler Taveira <[email protected]>; Amit Kapila <[email protected]>; pgsql-hackers; Bharath Rupireddy <[email protected]>
Dear Melih,
Thank you for updating the patch. Before reviewing, I found that
cfbot have not accepted v8 patch [1].
IIUC src/psql/describe.c has been modified in v8, but src/test/regress/expected/subscription.out
has not been changed accordingly.
[1]: https://cirrus-ci.com/github/postgresql-cfbot/postgresql/commitfest/42/3840
Best Regards,
Hayato Kuroda
FUJITSU LIMITED
^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2023-02-20 11:46 Melih Mutlu <[email protected]>
parent: Hayato Kuroda (Fujitsu) <[email protected]>
0 siblings, 4 replies; 14+ messages in thread
From: Melih Mutlu @ 2023-02-20 11:46 UTC (permalink / raw)
To: Hayato Kuroda (Fujitsu) <[email protected]>; +Cc: Takamichi Osumi (Fujitsu) <[email protected]>; [email protected] <[email protected]>; Euler Taveira <[email protected]>; Amit Kapila <[email protected]>; pgsql-hackers; Bharath Rupireddy <[email protected]>
Hi,
Hayato Kuroda (Fujitsu) <[email protected]>, 20 Şub 2023 Pzt, 10:12
tarihinde şunu yazdı:
> Dear Melih,
>
> Thank you for updating the patch. Before reviewing, I found that
> cfbot have not accepted v8 patch [1].
>
Thanks for letting me know.
Attached the fixed version of the patch.
Best,
--
Melih Mutlu
Microsoft
Attachments:
[application/octet-stream] v9-0001-Allow-logical-replication-to-copy-table-in-binary.patch (78.1K, ../../CAGPVpCTULGLwNGH7mMytR8dP1zg5f3wWq4=V=z=nT=-t8=2Vjg@mail.gmail.com/3-v9-0001-Allow-logical-replication-to-copy-table-in-binary.patch)
download | inline diff:
From 6327b4b54f4d17a15c5e29aa73ffbecf30d569fe Mon Sep 17 00:00:00 2001
From: Melih Mutlu <[email protected]>
Date: Mon, 8 Aug 2022 14:14:44 +0300
Subject: [PATCH v9] Allow logical replication to copy table in binary
If binary option is enabled in a subscription and copy_format is set to binary, then copy tables in
binary format during table synchronization.
The patch introduces a new parameter, copy_format, to CREATE SUBSCRIPTION to allow to choose
the format used in initial table synchronization.
Without this patch, table are copied in text format even if the
subscription is created with binary option. When binary format is
beneficial to use, allowing the subscription to copy tables in binary in
table sync phase may reduce the time spent on copy depending on column types.
Discussion: https://postgr.es/m/CAGPVpCQvAziCLknEnygY0v1-KBtg%2BOm-9JHJYZOnNPKFJPompw%40mail.gmail.com
---
doc/src/sgml/catalogs.sgml | 12 +-
doc/src/sgml/logical-replication.sgml | 9 +-
doc/src/sgml/ref/alter_subscription.sgml | 4 +-
doc/src/sgml/ref/create_subscription.sgml | 17 ++
src/backend/catalog/pg_subscription.c | 1 +
src/backend/commands/subscriptioncmds.c | 84 +++++-
src/backend/replication/logical/tablesync.c | 13 +-
src/bin/pg_dump/pg_dump.c | 17 +-
src/bin/pg_dump/pg_dump.h | 1 +
src/bin/psql/describe.c | 8 +-
src/bin/psql/tab-complete.c | 5 +-
src/include/catalog/pg_subscription.h | 13 +
src/include/commands/subscriptioncmds.h | 1 +
src/test/regress/expected/subscription.out | 192 ++++++++------
src/test/regress/sql/subscription.sql | 26 ++
src/test/subscription/t/002_types.pl | 269 +++++++++++++++-----
src/test/subscription/t/014_binary.pl | 82 +++++-
17 files changed, 604 insertions(+), 150 deletions(-)
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index c1e4048054..14e8cad531 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -7907,7 +7907,6 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
</para>
<para>
If true, the subscription will request that the publisher send data
- in binary format
</para></entry>
</row>
@@ -8002,6 +8001,17 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
origin.
</para></entry>
</row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>subcopyformat</structfield> <type>bool</type>
+ </para>
+ <para>
+ Specifies the format for copying the initial data in a subscription.
+ <literal>t</literal> indicates text, <literal>b</literal> indicates
+ binary format.
+ </para></entry>
+ </row>
</tbody>
</tgroup>
</table>
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 1bd5660c87..9540917ea3 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -241,10 +241,11 @@
types of the columns do not need to match, as long as the text
representation of the data can be converted to the target type. For
example, you can replicate from a column of type <type>integer</type> to a
- 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.
+ column of type <type>bigint</type>. Replication in binary format is type
+ specific and does not allow to replicate data between different types according
+ to its restrictions. 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.
</para>
<sect2 id="logical-replication-subscription-slot">
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 964fcbb8ff..c560d71f91 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -213,8 +213,8 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
are <literal>slot_name</literal>,
<literal>synchronous_commit</literal>,
<literal>binary</literal>, <literal>streaming</literal>,
- <literal>disable_on_error</literal>, and
- <literal>origin</literal>.
+ <literal>disable_on_error</literal>, <literal>origin</literal>
+ and <literal>copy_format</literal>.
</para>
</listitem>
</varlistentry>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 51c45f17c7..f01b762d81 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -349,6 +349,23 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
</para>
</listitem>
</varlistentry>
+
+ <varlistentry>
+ <term><literal>copy_format</literal> (<type>string</type>)</term>
+ <listitem>
+ <para>
+ Specifies the format in which pre-existing data on the publisher will
+ copied to the subscriber. Supported formats are
+ <literal>text</literal> and <literal>binary</literal>. The default is
+ <literal>text</literal>.
+ <literal>binary</literal> format can be selected only if
+ <literal>binary</literal> is set to <literal>true</literal>. Also,
+ the binary format is very data type specific, it will not allow copying
+ between different column types as opposed to text format. Note that
+ data will not be copied if <literal>copy_data = false</literal>.
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist></para>
</listitem>
diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index a56ae311c3..da078acefe 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -71,6 +71,7 @@ GetSubscription(Oid subid, bool missing_ok)
sub->stream = subform->substream;
sub->twophasestate = subform->subtwophasestate;
sub->disableonerr = subform->subdisableonerr;
+ sub->copyformat = subform->subcopyformat;
/* Get conninfo */
datum = SysCacheGetAttr(SUBSCRIPTIONOID,
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 464db6d247..4e4023623d 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -66,6 +66,7 @@
#define SUBOPT_DISABLE_ON_ERR 0x00000400
#define SUBOPT_LSN 0x00000800
#define SUBOPT_ORIGIN 0x00001000
+#define SUBOPT_COPY_FORMAT 0x00002000
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
@@ -90,6 +91,7 @@ typedef struct SubOpts
bool disableonerr;
char *origin;
XLogRecPtr lsn;
+ char copy_format;
} SubOpts;
static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
@@ -146,6 +148,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->disableonerr = false;
if (IsSet(supported_opts, SUBOPT_ORIGIN))
opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY);
+ if (IsSet(supported_opts, SUBOPT_COPY_FORMAT))
+ opts->copy_format = LOGICALREP_COPY_AS_TEXT;
/* Parse options */
foreach(lc, stmt_options)
@@ -324,6 +328,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->specified_opts |= SUBOPT_LSN;
opts->lsn = lsn;
}
+ else if (IsSet(supported_opts, SUBOPT_COPY_FORMAT) &&
+ strcmp(defel->defname, "copy_format") == 0)
+ {
+ if (IsSet(opts->specified_opts, SUBOPT_COPY_FORMAT))
+ errorConflictingDefElem(defel, pstate);
+
+ opts->specified_opts |= SUBOPT_COPY_FORMAT;
+ opts->copy_format = defGetCopyFormat(defel);
+ }
else
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
@@ -404,6 +417,17 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
"slot_name = NONE", "create_slot = false")));
}
}
+
+ if (IsSet(opts->specified_opts, SUBOPT_BINARY) &&
+ !opts->binary &&
+ opts->copy_format == LOGICALREP_COPY_AS_BINARY)
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("%s and %s are mutually exclusive options",
+ "binary = false", "copy_format = binary")));
+ }
+
}
/*
@@ -560,7 +584,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
SUBOPT_SLOT_NAME | SUBOPT_COPY_DATA |
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
- SUBOPT_DISABLE_ON_ERR | SUBOPT_ORIGIN);
+ SUBOPT_DISABLE_ON_ERR | SUBOPT_ORIGIN |
+ SUBOPT_COPY_FORMAT);
parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
/*
@@ -649,6 +674,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
publicationListToArray(publications);
values[Anum_pg_subscription_suborigin - 1] =
CStringGetTextDatum(opts.origin);
+ values[Anum_pg_subscription_subcopyformat - 1] = CharGetDatum(opts.copy_format);
tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
@@ -1054,7 +1080,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
supported_opts = (SUBOPT_SLOT_NAME |
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
SUBOPT_STREAMING | SUBOPT_DISABLE_ON_ERR |
- SUBOPT_ORIGIN);
+ SUBOPT_ORIGIN | SUBOPT_COPY_FORMAT);
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -1091,6 +1117,15 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
if (IsSet(opts.specified_opts, SUBOPT_BINARY))
{
+ /* Do not allow binary = false with copy_format = binary */
+ if (!opts.binary &&
+ sub->copyformat == LOGICALREP_COPY_AS_BINARY &&
+ !IsSet(opts.specified_opts, SUBOPT_COPY_FORMAT))
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot set %s for a subscription with %s",
+ "binary = false", "copy_format = binary")));
+
values[Anum_pg_subscription_subbinary - 1] =
BoolGetDatum(opts.binary);
replaces[Anum_pg_subscription_subbinary - 1] = true;
@@ -1118,6 +1153,21 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
replaces[Anum_pg_subscription_suborigin - 1] = true;
}
+ if (IsSet(opts.specified_opts, SUBOPT_COPY_FORMAT))
+ {
+ /* Do not allow binary = false with copy_format = binary */
+ if (opts.copy_format == LOGICALREP_COPY_AS_BINARY &&
+ !sub->binary &&
+ !IsSet(opts.specified_opts, SUBOPT_BINARY))
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot set %s for a subscription with %s",
+ "copy_format = binary", "binary = false")));
+ values[Anum_pg_subscription_subcopyformat - 1] =
+ CharGetDatum(opts.copy_format);
+ replaces[Anum_pg_subscription_subcopyformat - 1] = true;
+ }
+
update_tuple = true;
break;
}
@@ -2195,3 +2245,33 @@ defGetStreamingMode(DefElem *def)
def->defname)));
return LOGICALREP_STREAM_OFF; /* keep compiler quiet */
}
+
+/*
+ * Extract the copy format value from a DefElem.
+ */
+char
+defGetCopyFormat(DefElem *def)
+{
+ char *sval;
+
+ /*
+ * If no parameter value given, set it to text format.
+ */
+ if (!def->arg)
+ return LOGICALREP_COPY_AS_TEXT;
+
+ /*
+ * Currently supported formats are "text" and "binary".
+ */
+ sval = defGetString(def);
+ if (pg_strcasecmp(sval, "text") == 0)
+ return LOGICALREP_COPY_AS_TEXT;
+ if (pg_strcasecmp(sval, "binary") == 0)
+ return LOGICALREP_COPY_AS_BINARY;
+
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("%s value must be either \"text\" or \"binary\"",
+ def->defname)));
+ return LOGICALREP_COPY_AS_TEXT; /* keep compiler quiet */
+}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 07eea504ba..62c866c9eb 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -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"
@@ -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)),
@@ -1168,6 +1170,15 @@ copy_table(Relation rel)
appendStringInfoString(&cmd, ") TO STDOUT");
}
+
+ /* If the publisher is v16 or later, specify the format to copy data. */
+ if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 160000)
+ {
+ char *format = MySubscription->copyformat == LOGICALREP_COPY_AS_BINARY ? "binary" : "text";
+ appendStringInfo(&cmd, " WITH (FORMAT %s)", format);
+ options = lappend(options, makeDefElem("format", (Node *) makeString(format), -1));
+ }
+
res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 0, NULL);
pfree(cmd.data);
if (res->status != WALRCV_OK_COPY_OUT)
@@ -1184,7 +1195,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);
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 527c7651ab..9b97aa06e5 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4494,6 +4494,7 @@ getSubscriptions(Archive *fout)
int i_subsynccommit;
int i_subpublications;
int i_subbinary;
+ int i_subcopyformat;
int i,
ntups;
@@ -4546,9 +4547,15 @@ getSubscriptions(Archive *fout)
LOGICALREP_TWOPHASE_STATE_DISABLED);
if (fout->remoteVersion >= 160000)
- appendPQExpBufferStr(query, " s.suborigin\n");
+ {
+ appendPQExpBufferStr(query, " s.suborigin,\n");
+ appendPQExpBufferStr(query, " s.subcopyformat\n");
+ }
else
- appendPQExpBuffer(query, " '%s' AS suborigin\n", LOGICALREP_ORIGIN_ANY);
+ {
+ appendPQExpBuffer(query, " '%s' AS suborigin,\n", LOGICALREP_ORIGIN_ANY);
+ appendPQExpBuffer(query, " '%c' AS subcopyformat\n", LOGICALREP_COPY_AS_TEXT);
+ }
appendPQExpBufferStr(query,
"FROM pg_subscription s\n"
@@ -4576,6 +4583,7 @@ getSubscriptions(Archive *fout)
i_subtwophasestate = PQfnumber(res, "subtwophasestate");
i_subdisableonerr = PQfnumber(res, "subdisableonerr");
i_suborigin = PQfnumber(res, "suborigin");
+ i_subcopyformat = PQfnumber(res, "subcopyformat");
subinfo = pg_malloc(ntups * sizeof(SubscriptionInfo));
@@ -4606,6 +4614,8 @@ getSubscriptions(Archive *fout)
subinfo[i].subdisableonerr =
pg_strdup(PQgetvalue(res, i, i_subdisableonerr));
subinfo[i].suborigin = pg_strdup(PQgetvalue(res, i, i_suborigin));
+ subinfo[i].subcopyformat =
+ pg_strdup(PQgetvalue(res, i, i_subcopyformat));
/* Decide whether we want to dump it */
selectDumpableObject(&(subinfo[i].dobj), fout);
@@ -4687,6 +4697,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
if (strcmp(subinfo->subsynccommit, "off") != 0)
appendPQExpBuffer(query, ", synchronous_commit = %s", fmtId(subinfo->subsynccommit));
+ if (strcmp(subinfo->subcopyformat, "t") != 0)
+ appendPQExpBuffer(query, ", copy_format = %s", subinfo->subcopyformat);
+
appendPQExpBufferStr(query, ");\n");
if (subinfo->dobj.dump & DUMP_COMPONENT_DEFINITION)
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index e7cbd8d7ed..f6b7dbcd97 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -662,6 +662,7 @@ typedef struct _SubscriptionInfo
char *suborigin;
char *subsynccommit;
char *subpublications;
+ char *subcopyformat;
} SubscriptionInfo;
/*
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index c8a0bb7b3a..8e9523aba4 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -6472,7 +6472,7 @@ describeSubscriptions(const char *pattern, bool verbose)
PGresult *res;
printQueryOpt myopt = pset.popt;
static const bool translate_columns[] = {false, false, false, false,
- false, false, false, false, false, false, false, false};
+ false, false, false, false, false, false, false, false, false};
if (pset.sversion < 100000)
{
@@ -6528,9 +6528,15 @@ describeSubscriptions(const char *pattern, bool verbose)
gettext_noop("Disable on error"));
if (pset.sversion >= 160000)
+ {
appendPQExpBuffer(&buf,
", suborigin AS \"%s\"\n",
gettext_noop("Origin"));
+ /* Copy format is only supported in v16 and higher */
+ appendPQExpBuffer(&buf,
+ ", subcopyformat AS \"%s\"\n",
+ gettext_noop("Copy Format"));
+ }
appendPQExpBuffer(&buf,
", subsynccommit AS \"%s\"\n"
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 5e1882eaea..acc178780c 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1926,7 +1926,7 @@ psql_completion(const char *text, int start, int end)
/* ALTER SUBSCRIPTION <name> SET ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "("))
COMPLETE_WITH("binary", "disable_on_error", "origin", "slot_name",
- "streaming", "synchronous_commit");
+ "streaming", "synchronous_commit", "copy_format");
/* ALTER SUBSCRIPTION <name> SKIP ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SKIP", "("))
COMPLETE_WITH("lsn");
@@ -3269,7 +3269,8 @@ psql_completion(const char *text, int start, int end)
else if (HeadMatches("CREATE", "SUBSCRIPTION") && TailMatches("WITH", "("))
COMPLETE_WITH("binary", "connect", "copy_data", "create_slot",
"disable_on_error", "enabled", "origin", "slot_name",
- "streaming", "synchronous_commit", "two_phase");
+ "streaming", "synchronous_commit", "two_phase",
+ "copy_format");
/* CREATE TRIGGER --- is allowed inside CREATE SCHEMA, so use TailMatches */
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index b0f2a1705d..fe74cf2cce 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -88,6 +88,7 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
bool subdisableonerr; /* True if a worker error should cause the
* subscription to be disabled */
+ char subcopyformat BKI_DEFAULT(LOGICALREP_COPY_AS_TEXT); /* Copy format */
#ifdef CATALOG_VARLEN /* variable-length fields start here */
/* Connection string to the publisher */
text subconninfo BKI_FORCE_NOT_NULL;
@@ -137,6 +138,8 @@ typedef struct Subscription
List *publications; /* List of publication names to subscribe to */
char *origin; /* Only publish data originating from the
* specified origin */
+ char copyformat; /* Copy format for subscriptions with binary
+ * option enabled */
} Subscription;
/* Disallow streaming in-progress transactions. */
@@ -154,6 +157,16 @@ typedef struct Subscription
*/
#define LOGICALREP_STREAM_PARALLEL 'p'
+/*
+ * Copy data as text. This is the default.
+ */
+#define LOGICALREP_COPY_AS_TEXT 't'
+
+/*
+ * Copy data as binary if the binary options is enabled in the subscription.
+ */
+#define LOGICALREP_COPY_AS_BINARY 'b'
+
extern Subscription *GetSubscription(Oid subid, bool missing_ok);
extern void FreeSubscription(Subscription *sub);
extern void DisableSubscription(Oid subid);
diff --git a/src/include/commands/subscriptioncmds.h b/src/include/commands/subscriptioncmds.h
index 214dc6c29e..7937defea5 100644
--- a/src/include/commands/subscriptioncmds.h
+++ b/src/include/commands/subscriptioncmds.h
@@ -27,5 +27,6 @@ extern ObjectAddress AlterSubscriptionOwner(const char *name, Oid newOwnerId);
extern void AlterSubscriptionOwner_oid(Oid subid, Oid newOwnerId);
extern char defGetStreamingMode(DefElem *def);
+extern char defGetCopyFormat(DefElem *def);
#endif /* SUBSCRIPTIONCMDS_H */
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 3f99b14394..e80666db99 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -114,18 +114,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | none | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | none | t | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub4 SET (origin = any);
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub3;
@@ -143,10 +143,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@@ -163,10 +163,10 @@ ERROR: unrecognized subscription parameter: "create_slot"
-- ok
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345');
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | off | dbname=regress_doesnotexist2 | 0/12345
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | off | dbname=regress_doesnotexist2 | 0/12345
(1 row)
-- ok - with lsn = NONE
@@ -175,10 +175,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE);
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0');
ERROR: invalid WAL location (LSN): 0/0
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | off | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | off | dbname=regress_doesnotexist2 | 0/0
(1 row)
BEGIN;
@@ -210,10 +210,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar);
ERROR: invalid value for parameter "synchronous_commit": "foobar"
HINT: Available values: local, remote_write, remote_apply, on, off.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+--------------------+------------------------------+----------
- regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | local | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+------------------------------+----------
+ regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | local | dbname=regress_doesnotexist2 | 0/0
(1 row)
-- rename back to keep the rest simple
@@ -247,19 +247,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | t | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (binary = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -271,27 +271,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication already exists
@@ -306,10 +306,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr
ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false);
ERROR: publication "testpub1" is already in subscription "regress_testsub"
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication used more than once
@@ -324,10 +324,10 @@ ERROR: publication "testpub3" is not in subscription "regress_testsub"
-- ok - delete publications
ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -363,10 +363,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | t | off | dbname=regress_doesnotexist | 0/0
(1 row)
--fail - alter of two_phase option not supported.
@@ -375,10 +375,10 @@ ERROR: unrecognized subscription parameter: "two_phase"
-- but can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -388,10 +388,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -404,22 +404,70 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | t | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | t | any | t | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
DROP SUBSCRIPTION regress_testsub;
+-- fail - cannot set binary = false and copy_format = binary in a subscription
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (binary = false, copy_format = binary, connect = false, slot_name = NONE);
+ERROR: binary = false and copy_format = binary are mutually exclusive options
+-- now it works
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (binary = true, copy_format = binary, connect = false, slot_name = NONE);
+WARNING: subscription was created, but is not connected
+HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
+\dRs+
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | b | off | dbname=regress_doesnotexist | 0/0
+(1 row)
+
+-- alter copy_format to text
+ALTER SUBSCRIPTION regress_testsub SET (copy_format = text);
+\dRs+
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | t | off | dbname=regress_doesnotexist | 0/0
+(1 row)
+
+-- alter copy_format back to binary
+ALTER SUBSCRIPTION regress_testsub SET (copy_format = binary);
+\dRs+
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | b | off | dbname=regress_doesnotexist | 0/0
+(1 row)
+
+-- fail - cannot set binary = false if copy_format = binary
+ALTER SUBSCRIPTION regress_testsub SET (binary = false);
+ERROR: cannot set binary = false for a subscription with copy_format = binary
+-- now it works
+ALTER SUBSCRIPTION regress_testsub SET (binary = false, copy_format = text);
+\dRs+
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy Format | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | off | dbname=regress_doesnotexist | 0/0
+(1 row)
+
+-- fail - cannot set copy_format = binary for a subscription with binary = false
+ALTER SUBSCRIPTION regress_testsub SET (copy_format = binary);
+ERROR: cannot set copy_format = binary for a subscription with binary = false
+DROP SUBSCRIPTION regress_testsub;
RESET SESSION AUTHORIZATION;
DROP ROLE regress_subscription_user;
DROP ROLE regress_subscription_user2;
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index 7281f5fee2..875f3af6bf 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -286,6 +286,32 @@ ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
DROP SUBSCRIPTION regress_testsub;
+-- fail - cannot set binary = false and copy_format = binary in a subscription
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (binary = false, copy_format = binary, connect = false, slot_name = NONE);
+
+-- now it works
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (binary = true, copy_format = binary, connect = false, slot_name = NONE);
+\dRs+
+
+-- alter copy_format to text
+ALTER SUBSCRIPTION regress_testsub SET (copy_format = text);
+\dRs+
+
+-- alter copy_format back to binary
+ALTER SUBSCRIPTION regress_testsub SET (copy_format = binary);
+\dRs+
+
+-- fail - cannot set binary = false if copy_format = binary
+ALTER SUBSCRIPTION regress_testsub SET (binary = false);
+
+-- now it works
+ALTER SUBSCRIPTION regress_testsub SET (binary = false, copy_format = text);
+\dRs+
+
+-- fail - cannot set copy_format = binary for a subscription with binary = false
+ALTER SUBSCRIPTION regress_testsub SET (copy_format = binary);
+DROP SUBSCRIPTION regress_testsub;
+
RESET SESSION AUTHORIZATION;
DROP ROLE regress_subscription_user;
DROP ROLE regress_subscription_user2;
diff --git a/src/test/subscription/t/002_types.pl b/src/test/subscription/t/002_types.pl
index 6b5853b80b..02db58dc3b 100644
--- a/src/test/subscription/t/002_types.pl
+++ b/src/test/subscription/t/002_types.pl
@@ -19,6 +19,11 @@ my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
$node_subscriber->init(allows_streaming => 'logical');
$node_subscriber->start;
+# Create binary subscriber node
+my $node_subscriber_binary = PostgreSQL::Test::Cluster->new('subscriber_binary');
+$node_subscriber_binary->init(allows_streaming => 'logical');
+$node_subscriber_binary->start;
+
# Create some preexisting content on publisher
my $ddl = qq(
CREATE EXTENSION hstore WITH SCHEMA public;
@@ -104,6 +109,100 @@ my $ddl = qq(
# Setup structure on both nodes
$node_publisher->safe_psql('postgres', $ddl);
$node_subscriber->safe_psql('postgres', $ddl);
+$node_subscriber_binary->safe_psql('postgres', $ddl);
+
+# Insert initial test data
+$node_publisher->safe_psql(
+ 'postgres', qq(
+ -- test_tbl_one_array_col
+ INSERT INTO tst_one_array (a, b) VALUES
+ (1, '{1, 2, 3}'),
+ (2, '{2, 3, 1}');
+
+ -- test_tbl_arrays
+ INSERT INTO tst_arrays (a, b, c, d) VALUES
+ ('{1, 2, 3}', '{"a", "b", "c"}', '{1.1, 2.2, 3.3}', '{"1 day", "2 days", "3 days"}'),
+ ('{2, 3, 1}', '{"b", "c", "a"}', '{2.2, 3.3, 1.1}', '{"2 minutes", "3 minutes", "1 minute"}');
+
+ -- test_tbl_single_enum
+ INSERT INTO tst_one_enum (a, b) VALUES
+ (1, 'a'),
+ (2, 'b');
+
+ -- test_tbl_enums
+ INSERT INTO tst_enums (a, b) VALUES
+ ('a', '{b, c}'),
+ ('b', '{c, a}');
+
+ -- test_tbl_single_composites
+ INSERT INTO tst_one_comp (a, b) VALUES
+ (1, ROW(1.0, 'a', 1)),
+ (2, ROW(2.0, 'b', 2));
+
+ -- test_tbl_composites
+ INSERT INTO tst_comps (a, b) VALUES
+ (ROW(1.0, 'a', 1), ARRAY[ROW(1, 'a', 1)::tst_comp_basic_t]),
+ (ROW(2.0, 'b', 2), ARRAY[ROW(2, 'b', 2)::tst_comp_basic_t]);
+
+ -- test_tbl_composite_with_enums
+ INSERT INTO tst_comp_enum (a, b) VALUES
+ (1, ROW(1.0, 'a', 1)),
+ (2, ROW(2.0, 'b', 2));
+
+ -- test_tbl_composite_with_enums_array
+ INSERT INTO tst_comp_enum_array (a, b) VALUES
+ (ROW(1.0, 'a', 1), ARRAY[ROW(1, 'a', 1)::tst_comp_enum_t]),
+ (ROW(2.0, 'b', 2), ARRAY[ROW(2, 'b', 2)::tst_comp_enum_t]);
+
+ -- test_tbl_composite_with_single_enums_array_in_composite
+ INSERT INTO tst_comp_one_enum_array (a, b) VALUES
+ (1, ROW(1.0, '{a, b, c}', 1)),
+ (2, ROW(2.0, '{a, b, c}', 2));
+
+ -- test_tbl_composite_with_enums_array_in_composite
+ INSERT INTO tst_comp_enum_what (a, b) VALUES
+ (ROW(1.0, '{a, b, c}', 1), ARRAY[ROW(1, '{a, b, c}', 1)::tst_comp_enum_array_t]),
+ (ROW(2.0, '{b, c, a}', 2), ARRAY[ROW(2, '{b, c, a}', 1)::tst_comp_enum_array_t]);
+
+ -- test_tbl_mixed_composites
+ INSERT INTO tst_comp_mix_array (a, b) VALUES
+ (ROW(
+ ROW(1,'a',1),
+ ARRAY[ROW(1,'a',1)::tst_comp_basic_t, ROW(2,'b',2)::tst_comp_basic_t],
+ 'a',
+ '{a,b,NULL,c}'),
+ ARRAY[
+ ROW(
+ ROW(1,'a',1),
+ ARRAY[
+ ROW(1,'a',1)::tst_comp_basic_t,
+ ROW(2,'b',2)::tst_comp_basic_t,
+ NULL
+ ],
+ 'a',
+ '{a,b,c}'
+ )::tst_comp_mix_t
+ ]
+ );
+
+ -- test_tbl_range
+ INSERT INTO tst_range (a, b) VALUES
+ (1, '[1, 10]'),
+ (2, '[2, 20]');
+
+ -- test_tbl_range_array
+ INSERT INTO tst_range_array (a, b, c) VALUES
+ (1, tstzrange('Mon Aug 04 00:00:00 2014 CEST'::timestamptz, 'infinity'), '{"[1,2]", "[10,20]"}'),
+ (2, tstzrange('Sat Aug 02 00:00:00 2014 CEST'::timestamptz, 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[2,3]", "[20,30]"}');
+
+ -- tst_hstore
+ INSERT INTO tst_hstore (a, b) VALUES
+ (1, '"a"=>"1"'),
+ (2, '"zzz"=>"foo"');
+
+ -- tst_dom_constr
+ INSERT INTO tst_dom_constr VALUES (10);
+));
# Setup logical replication
my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
@@ -113,89 +212,126 @@ $node_publisher->safe_psql('postgres',
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr' PUBLICATION tap_pub WITH (slot_name = tap_sub_slot)"
);
+$node_subscriber_binary->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub_binary CONNECTION '$publisher_connstr' PUBLICATION tap_pub WITH (slot_name = tap_sub_binary_slot, binary = true, copy_format = 'binary')"
+);
# Wait for initial sync to finish as well
$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub');
+$node_subscriber_binary->wait_for_subscription_sync($node_publisher, 'tap_sub_binary');
-# Insert initial test data
+my $sync_check = qq(
+ SET timezone = '+2';
+ SELECT a, b FROM tst_one_array ORDER BY a;
+ SELECT a, b, c, d FROM tst_arrays ORDER BY a;
+ SELECT a, b FROM tst_one_enum ORDER BY a;
+ SELECT a, b FROM tst_enums ORDER BY a;
+ SELECT a, b FROM tst_one_comp ORDER BY a;
+ SELECT a, b FROM tst_comps ORDER BY a;
+ SELECT a, b FROM tst_comp_enum ORDER BY a;
+ SELECT a, b FROM tst_comp_enum_array ORDER BY a;
+ SELECT a, b FROM tst_comp_one_enum_array ORDER BY a;
+ SELECT a, b FROM tst_comp_enum_what ORDER BY a;
+ SELECT a, b FROM tst_comp_mix_array ORDER BY a;
+ SELECT a, b FROM tst_range ORDER BY a;
+ SELECT a, b, c FROM tst_range_array ORDER BY a;
+ SELECT a, b FROM tst_hstore ORDER BY a;
+);
+
+# Check the synced data on subscribers
+my $result = $node_subscriber->safe_psql('postgres', $sync_check);
+my $result_binary = $node_subscriber_binary->safe_psql('postgres', $sync_check);
+
+my $sync_result = '1|{1,2,3}
+2|{2,3,1}
+{1,2,3}|{a,b,c}|{1.1,2.2,3.3}|{"1 day","2 days","3 days"}
+{2,3,1}|{b,c,a}|{2.2,3.3,1.1}|{00:02:00,00:03:00,00:01:00}
+1|a
+2|b
+a|{b,c}
+b|{c,a}
+1|(1,a,1)
+2|(2,b,2)
+(1,a,1)|{"(1,a,1)"}
+(2,b,2)|{"(2,b,2)"}
+1|(1,a,1)
+2|(2,b,2)
+(1,a,1)|{"(1,a,1)"}
+(2,b,2)|{"(2,b,2)"}
+1|(1,"{a,b,c}",1)
+2|(2,"{a,b,c}",2)
+(1,"{a,b,c}",1)|{"(1,\"{a,b,c}\",1)"}
+(2,"{b,c,a}",2)|{"(2,\"{b,c,a}\",1)"}
+("(1,a,1)","{""(1,a,1)"",""(2,b,2)""}",a,"{a,b,NULL,c}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
+1|[1,11)
+2|[2,21)
+1|["2014-08-04 00:00:00+02",infinity)|{"[1,3)","[10,21)"}
+2|["2014-08-02 00:00:00+02","2014-08-04 00:00:00+02")|{"[2,4)","[20,31)"}
+1|"a"=>"1"
+2|"zzz"=>"foo"';
+
+is( $result, $sync_result, 'check initial sync on subscriber');
+is( $result_binary, $sync_result, 'check initial sync on subscriber in binary');
+
+# Insert test data for update to apply changes
$node_publisher->safe_psql(
'postgres', qq(
-- test_tbl_one_array_col
INSERT INTO tst_one_array (a, b) VALUES
- (1, '{1, 2, 3}'),
- (2, '{2, 3, 1}'),
(3, '{3, 2, 1}'),
(4, '{4, 3, 2}'),
(5, '{5, NULL, 3}');
-- test_tbl_arrays
INSERT INTO tst_arrays (a, b, c, d) VALUES
- ('{1, 2, 3}', '{"a", "b", "c"}', '{1.1, 2.2, 3.3}', '{"1 day", "2 days", "3 days"}'),
- ('{2, 3, 1}', '{"b", "c", "a"}', '{2.2, 3.3, 1.1}', '{"2 minutes", "3 minutes", "1 minute"}'),
('{3, 1, 2}', '{"c", "a", "b"}', '{3.3, 1.1, 2.2}', '{"3 years", "1 year", "2 years"}'),
('{4, 1, 2}', '{"d", "a", "b"}', '{4.4, 1.1, 2.2}', '{"4 years", "1 year", "2 years"}'),
('{5, NULL, NULL}', '{"e", NULL, "b"}', '{5.5, 1.1, NULL}', '{"5 years", NULL, NULL}');
-- test_tbl_single_enum
INSERT INTO tst_one_enum (a, b) VALUES
- (1, 'a'),
- (2, 'b'),
(3, 'c'),
(4, 'd'),
(5, NULL);
-- test_tbl_enums
INSERT INTO tst_enums (a, b) VALUES
- ('a', '{b, c}'),
- ('b', '{c, a}'),
('c', '{b, a}'),
('d', '{c, b}'),
('e', '{d, NULL}');
-- test_tbl_single_composites
INSERT INTO tst_one_comp (a, b) VALUES
- (1, ROW(1.0, 'a', 1)),
- (2, ROW(2.0, 'b', 2)),
(3, ROW(3.0, 'c', 3)),
(4, ROW(4.0, 'd', 4)),
(5, ROW(NULL, NULL, 5));
-- test_tbl_composites
INSERT INTO tst_comps (a, b) VALUES
- (ROW(1.0, 'a', 1), ARRAY[ROW(1, 'a', 1)::tst_comp_basic_t]),
- (ROW(2.0, 'b', 2), ARRAY[ROW(2, 'b', 2)::tst_comp_basic_t]),
(ROW(3.0, 'c', 3), ARRAY[ROW(3, 'c', 3)::tst_comp_basic_t]),
(ROW(4.0, 'd', 4), ARRAY[ROW(4, 'd', 3)::tst_comp_basic_t]),
(ROW(5.0, 'e', NULL), ARRAY[NULL, ROW(5, NULL, 5)::tst_comp_basic_t]);
-- test_tbl_composite_with_enums
INSERT INTO tst_comp_enum (a, b) VALUES
- (1, ROW(1.0, 'a', 1)),
- (2, ROW(2.0, 'b', 2)),
(3, ROW(3.0, 'c', 3)),
(4, ROW(4.0, 'd', 4)),
(5, ROW(NULL, 'e', NULL));
-- test_tbl_composite_with_enums_array
INSERT INTO tst_comp_enum_array (a, b) VALUES
- (ROW(1.0, 'a', 1), ARRAY[ROW(1, 'a', 1)::tst_comp_enum_t]),
- (ROW(2.0, 'b', 2), ARRAY[ROW(2, 'b', 2)::tst_comp_enum_t]),
(ROW(3.0, 'c', 3), ARRAY[ROW(3, 'c', 3)::tst_comp_enum_t]),
(ROW(4.0, 'd', 3), ARRAY[ROW(3, 'd', 3)::tst_comp_enum_t]),
(ROW(5.0, 'e', 3), ARRAY[ROW(3, 'e', 3)::tst_comp_enum_t, NULL]);
-- test_tbl_composite_with_single_enums_array_in_composite
INSERT INTO tst_comp_one_enum_array (a, b) VALUES
- (1, ROW(1.0, '{a, b, c}', 1)),
- (2, ROW(2.0, '{a, b, c}', 2)),
(3, ROW(3.0, '{a, b, c}', 3)),
(4, ROW(4.0, '{c, b, d}', 4)),
(5, ROW(5.0, '{NULL, e, NULL}', 5));
-- test_tbl_composite_with_enums_array_in_composite
INSERT INTO tst_comp_enum_what (a, b) VALUES
- (ROW(1.0, '{a, b, c}', 1), ARRAY[ROW(1, '{a, b, c}', 1)::tst_comp_enum_array_t]),
- (ROW(2.0, '{b, c, a}', 2), ARRAY[ROW(2, '{b, c, a}', 1)::tst_comp_enum_array_t]),
(ROW(3.0, '{c, a, b}', 1), ARRAY[ROW(3, '{c, a, b}', 1)::tst_comp_enum_array_t]),
(ROW(4.0, '{c, b, d}', 4), ARRAY[ROW(4, '{c, b, d}', 4)::tst_comp_enum_array_t]),
(ROW(5.0, '{c, NULL, b}', NULL), ARRAY[ROW(5, '{c, e, b}', 1)::tst_comp_enum_array_t]);
@@ -203,10 +339,10 @@ $node_publisher->safe_psql(
-- test_tbl_mixed_composites
INSERT INTO tst_comp_mix_array (a, b) VALUES
(ROW(
- ROW(1,'a',1),
- ARRAY[ROW(1,'a',1)::tst_comp_basic_t, ROW(2,'b',2)::tst_comp_basic_t],
- 'a',
- '{a,b,NULL,c}'),
+ ROW(2,'b',2),
+ ARRAY[ROW(2,'b',2)::tst_comp_basic_t, ROW(3,'c',3)::tst_comp_basic_t],
+ 'b',
+ '{b,c,NULL,d}'),
ARRAY[
ROW(
ROW(1,'a',1),
@@ -223,36 +359,29 @@ $node_publisher->safe_psql(
-- test_tbl_range
INSERT INTO tst_range (a, b) VALUES
- (1, '[1, 10]'),
- (2, '[2, 20]'),
(3, '[3, 30]'),
(4, '[4, 40]'),
(5, '[5, 50]');
-- test_tbl_range_array
INSERT INTO tst_range_array (a, b, c) VALUES
- (1, tstzrange('Mon Aug 04 00:00:00 2014 CEST'::timestamptz, 'infinity'), '{"[1,2]", "[10,20]"}'),
- (2, tstzrange('Sat Aug 02 00:00:00 2014 CEST'::timestamptz, 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[2,3]", "[20,30]"}'),
(3, tstzrange('Fri Aug 01 00:00:00 2014 CEST'::timestamptz, 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[3,4]"}'),
(4, tstzrange('Thu Jul 31 00:00:00 2014 CEST'::timestamptz, 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[4,5]", NULL, "[40,50]"}'),
(5, NULL, NULL);
-- tst_hstore
INSERT INTO tst_hstore (a, b) VALUES
- (1, '"a"=>"1"'),
- (2, '"zzz"=>"foo"'),
(3, '"123"=>"321"'),
(4, '"yellow horse"=>"moaned"');
-- tst_dom_constr
- INSERT INTO tst_dom_constr VALUES (10);
+ INSERT INTO tst_dom_constr VALUES (11);
));
$node_publisher->wait_for_catchup('tap_sub');
+$node_publisher->wait_for_catchup('tap_sub_binary');
-# Check the data on subscriber
-my $result = $node_subscriber->safe_psql(
- 'postgres', qq(
+my $initial_check = qq(
SET timezone = '+2';
SELECT a, b FROM tst_one_array ORDER BY a;
SELECT a, b, c, d FROM tst_arrays ORDER BY a;
@@ -268,9 +397,13 @@ my $result = $node_subscriber->safe_psql(
SELECT a, b FROM tst_range ORDER BY a;
SELECT a, b, c FROM tst_range_array ORDER BY a;
SELECT a, b FROM tst_hstore ORDER BY a;
-));
+);
+
+# Check the data on subscribers
+$result = $node_subscriber->safe_psql('postgres', $initial_check);
+$result_binary = $node_subscriber_binary->safe_psql('postgres', $initial_check);
-is( $result, '1|{1,2,3}
+my $initial_result = '1|{1,2,3}
2|{2,3,1}
3|{3,2,1}
4|{4,3,2}
@@ -321,6 +454,7 @@ e|{d,NULL}
(4,"{c,b,d}",4)|{"(4,\"{c,b,d}\",4)"}
(5,"{c,NULL,b}",)|{"(5,\"{c,e,b}\",1)"}
("(1,a,1)","{""(1,a,1)"",""(2,b,2)""}",a,"{a,b,NULL,c}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
+("(2,b,2)","{""(2,b,2)"",""(3,c,3)""}",b,"{b,c,NULL,d}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
1|[1,11)
2|[2,21)
3|[3,31)
@@ -334,8 +468,10 @@ e|{d,NULL}
1|"a"=>"1"
2|"zzz"=>"foo"
3|"123"=>"321"
-4|"yellow horse"=>"moaned"',
- 'check replicated inserts on subscriber');
+4|"yellow horse"=>"moaned"';
+
+is( $result, $initial_result, 'check replicated inserts on subscriber');
+is( $result_binary, $initial_result, 'check replicated inserts on subscriber in binary');
# Run batch of updates
$node_publisher->safe_psql(
@@ -370,10 +506,9 @@ $node_publisher->safe_psql(
));
$node_publisher->wait_for_catchup('tap_sub');
+$node_publisher->wait_for_catchup('tap_sub_binary');
-# Check the data on subscriber
-$result = $node_subscriber->safe_psql(
- 'postgres', qq(
+my $update_check = qq(
SET timezone = '+2';
SELECT a, b FROM tst_one_array ORDER BY a;
SELECT a, b, c, d FROM tst_arrays ORDER BY a;
@@ -389,9 +524,13 @@ $result = $node_subscriber->safe_psql(
SELECT a, b FROM tst_range ORDER BY a;
SELECT a, b, c FROM tst_range_array ORDER BY a;
SELECT a, b FROM tst_hstore ORDER BY a;
-));
+);
+
+# Check the data on subscribers
+$result = $node_subscriber->safe_psql('postgres', $update_check);
+$result_binary = $node_subscriber_binary->safe_psql('postgres', $update_check);
-is( $result, '1|{4,5,6}
+my $update_result = '1|{4,5,6}
2|{2,3,1}
3|{3,2,1}
4|{4,5,6,1}
@@ -442,6 +581,7 @@ e|{e,d}
(4,"{c,b,d}",4)|{"(5,\"{a,b,c}\",5)"}
(5,"{c,NULL,b}",)|{"(5,\"{a,b,c}\",5)"}
("(1,a,1)","{""(1,a,1)"",""(2,b,2)""}",a,"{a,b,NULL,c}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")",NULL}
+("(2,b,2)","{""(2,b,2)"",""(3,c,3)""}",b,"{b,c,NULL,d}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
1|[100,1001)
2|[2,21)
3|[3,31)
@@ -455,8 +595,10 @@ e|{e,d}
1|"updated"=>"value"
2|"updated"=>"value"
3|"also"=>"updated"
-4|"yellow horse"=>"moaned"',
- 'check replicated updates on subscriber');
+4|"yellow horse"=>"moaned"';
+
+is( $result, $update_result, 'check replicated updates on subscriber');
+is( $result_binary, $update_result, 'check replicated updates on subscriber in binary');
# Run batch of deletes
$node_publisher->safe_psql(
@@ -490,10 +632,9 @@ $node_publisher->safe_psql(
));
$node_publisher->wait_for_catchup('tap_sub');
+$node_publisher->wait_for_catchup('tap_sub_binary');
-# Check the data on subscriber
-$result = $node_subscriber->safe_psql(
- 'postgres', qq(
+my $delete_check = qq(
SET timezone = '+2';
SELECT a, b FROM tst_one_array ORDER BY a;
SELECT a, b, c, d FROM tst_arrays ORDER BY a;
@@ -509,9 +650,13 @@ $result = $node_subscriber->safe_psql(
SELECT a, b FROM tst_range ORDER BY a;
SELECT a, b, c FROM tst_range_array ORDER BY a;
SELECT a, b FROM tst_hstore ORDER BY a;
-));
+);
+
+# Check the data on subscribers
+$result = $node_subscriber->safe_psql('postgres', $delete_check);
+$result_binary = $node_subscriber_binary->safe_psql('postgres', $delete_check);
-is( $result, '3|{3,2,1}
+my $delete_result = '3|{3,2,1}
4|{4,5,6,1}
5|{4,5,6,1}
{3,1,2}|{c,a,b}|{3.3,1.1,2.2}|{"3 years","1 year","2 years"}
@@ -540,26 +685,34 @@ e|{e,d}
(2,"{b,c,a}",2)|{"(2,\"{b,c,a}\",1)"}
(4,"{c,b,d}",4)|{"(5,\"{a,b,c}\",5)"}
(5,"{c,NULL,b}",)|{"(5,\"{a,b,c}\",5)"}
+("(2,b,2)","{""(2,b,2)"",""(3,c,3)""}",b,"{b,c,NULL,d}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
2|["2014-08-02 00:00:00+02","2014-08-04 00:00:00+02")|{"[2,4)","[20,31)"}
3|["2014-08-01 00:00:00+02","2014-08-04 00:00:00+02")|{"[3,5)"}
2|"updated"=>"value"
3|"also"=>"updated"
-4|"yellow horse"=>"moaned"',
- 'check replicated deletes on subscriber');
+4|"yellow horse"=>"moaned"';
+
+is( $result, $delete_result, 'check replicated deletes on subscriber');
+is( $result_binary, $delete_result, 'check replicated deletes on subscriber in binary');
# Test a domain with a constraint backed by a SQL-language function,
# which needs an active snapshot in order to operate.
$node_publisher->safe_psql('postgres',
- "INSERT INTO tst_dom_constr VALUES (11)");
+ "INSERT INTO tst_dom_constr VALUES (12)");
$node_publisher->wait_for_catchup('tap_sub');
+$node_publisher->wait_for_catchup('tap_sub_binary');
+
+my $domain_check = qq(SELECT sum(a) FROM tst_dom_constr);
+
+$result = $node_subscriber->safe_psql('postgres', $domain_check);
+$result_binary = $node_subscriber->safe_psql('postgres', $domain_check);
-$result =
- $node_subscriber->safe_psql('postgres',
- "SELECT sum(a) FROM tst_dom_constr");
-is($result, '21', 'sql-function constraint on domain');
+is( $result, '33', 'sql-function constraint on domain');
+is( $result_binary, '33', 'sql-function constraint on domain');
$node_subscriber->stop('fast');
+$node_subscriber_binary->stop('fast');
$node_publisher->stop('fast');
done_testing();
diff --git a/src/test/subscription/t/014_binary.pl b/src/test/subscription/t/014_binary.pl
index e53e23da3e..4fee421733 100644
--- a/src/test/subscription/t/014_binary.pl
+++ b/src/test/subscription/t/014_binary.pl
@@ -36,6 +36,16 @@ my $ddl = qq(
$node_publisher->safe_psql('postgres', $ddl);
$node_subscriber->safe_psql('postgres', $ddl);
+# Insert some content and make sure it's synced to subscriber
+$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"}');
+
+ INSERT INTO public.test_numerical (a, b, c, d) VALUES
+ (1, 1.2, 1.3, 10);
+ ));
+
# Configure logical replication
$node_publisher->safe_psql('postgres',
"CREATE PUBLICATION tpub FOR ALL TABLES");
@@ -48,27 +58,35 @@ $node_subscriber->safe_psql('postgres',
# Ensure nodes are in sync with each other
$node_subscriber->wait_for_subscription_sync($node_publisher, 'tsub');
+my $result = $node_subscriber->safe_psql('postgres',
+ "SELECT a, b, c, d FROM test_numerical ORDER BY a;
+ SELECT a, b, c FROM test_arrays ORDER BY a;");
+
+is( $result, '1|1.2|1.3|10
+{1,2,3}|{1.1,1.2,1.3}|{one,two,three}', 'check syned data on subscriber');
+
# 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"}');
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);
));
$node_publisher->wait_for_catchup('tsub');
-my $result = $node_subscriber->safe_psql('postgres',
- "SELECT a, b, c, d FROM test_numerical ORDER BY a");
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT a, b, c, d FROM test_numerical ORDER BY a;
+ SELECT a, b, c FROM test_arrays 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
+{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 replicated data on subscriber');
# Test updates as well
$node_publisher->safe_psql(
@@ -130,6 +148,60 @@ is( $result, '{1,2,3}|{42,1.2,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 whether subscriber fails in case of column type mismatch
+$ddl = qq(
+ CREATE TABLE public.test_cross_type (a SMALLINT);
+ INSERT INTO public.test_cross_type VALUES (1);
+);
+$node_publisher->safe_psql('postgres', $ddl);
+
+$ddl = qq(
+ CREATE TABLE public.test_cross_type (a INTEGER);
+);
+$node_subscriber->safe_psql('postgres', $ddl);
+
+# Change the format of inital copy to binary
+$node_subscriber->safe_psql('postgres', 'ALTER SUBSCRIPTION tsub SET (copy_format = binary)');
+$node_subscriber->safe_psql('postgres', 'ALTER SUBSCRIPTION tsub REFRESH PUBLICATION');
+
+# Cannot copy initial data in binary format because of type mismatch
+$node_subscriber->wait_for_log(qr/ERROR: ( [A-Z0-9]+:)? insufficient data left in message/);
+
+# Reset the format of initial copy back to text
+$node_subscriber->safe_psql('postgres', 'ALTER SUBSCRIPTION tsub SET (copy_format = text)');
+
+# It should be able to sync the table now
+$node_subscriber->wait_for_subscription_sync($node_publisher,'tsub');
+
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT a FROM test_cross_type ORDER BY a");
+
+is( $result, '1', 'check initial sync of data with type mismatch');
+
+# Change the format of inital copy to binary again to test the apply phase
+$node_subscriber->safe_psql('postgres', 'ALTER SUBSCRIPTION tsub SET (copy_format = binary)');
+
+# Insert a new row
+$node_publisher->safe_psql(
+ 'postgres', qq(
+ INSERT INTO public.test_cross_type (a) VALUES
+ (2);
+ ));
+
+# Cannot apply due to type mismatch
+$node_subscriber->wait_for_log(qr/ERROR: ( [A-Z0-9]+:)? insufficient data left in message/);
+
+# Set binary =false and copy_format = text
+$node_subscriber->safe_psql('postgres', 'ALTER SUBSCRIPTION tsub SET (binary = false, copy_format = text)');
+
+$node_publisher->wait_for_catchup('tsub');
+
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT a FROM public.test_cross_type ORDER BY a");
+
+is( $result, '1
+2', 'check replication of data with type mismatch');
+
$node_subscriber->stop('fast');
$node_publisher->stop('fast');
--
2.25.1
^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2023-02-21 13:48 Bharath Rupireddy <[email protected]>
parent: Melih Mutlu <[email protected]>
3 siblings, 1 reply; 14+ messages in thread
From: Bharath Rupireddy @ 2023-02-21 13:48 UTC (permalink / raw)
To: Melih Mutlu <[email protected]>; +Cc: Hayato Kuroda (Fujitsu) <[email protected]>; Takamichi Osumi (Fujitsu) <[email protected]>; [email protected] <[email protected]>; Euler Taveira <[email protected]>; Amit Kapila <[email protected]>; pgsql-hackers
On Mon, Feb 20, 2023 at 5:17 PM Melih Mutlu <[email protected]> wrote:
>
> Thanks for letting me know.
> Attached the fixed version of the patch.
Thanks. I have few comments on v9 patch:
1.
+ /* Do not allow binary = false with copy_format = binary */
+ if (!opts.binary &&
+ sub->copyformat == LOGICALREP_COPY_AS_BINARY &&
+ !IsSet(opts.specified_opts, SUBOPT_COPY_FORMAT))
+ ereport(ERROR,
+
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot set %s for a
subscription with %s",
+ "binary = false",
"copy_format = binary")));
I don't understand why we'd need to tie an option (binary) that deals
with data types at column-level with another option (copy_format) that
requests the entire table data to be in binary. This'd essentially
make one to set binary = true to use copy_format = binary, no? IMHO,
this inter-dependency is not good for better usability.
2. Why can't the tests that this patch adds be simple? Why would it
need to change the existing tests at all? I'm thinking to create a new
00X_binary_copy_format.pl or such and setting up logical replication
with copy_format = binary and letting table sync worker request
publisher in binary format - you can verify this via publisher server
logs - look for COPY with BINARY option. If required, have the table
with different data types. This greatly reduces the patch's footprint.
--
Bharath Rupireddy
PostgreSQL Contributors Team
RDS Open Source Databases
Amazon Web Services: https://aws.amazon.com
^ permalink raw reply [nested|flat] 14+ messages in thread
* RE: Allow logical replication to copy tables in binary format
@ 2023-02-22 07:01 Hayato Kuroda (Fujitsu) <[email protected]>
parent: Melih Mutlu <[email protected]>
3 siblings, 0 replies; 14+ messages in thread
From: Hayato Kuroda (Fujitsu) @ 2023-02-22 07:01 UTC (permalink / raw)
To: 'Melih Mutlu' <[email protected]>; +Cc: Takamichi Osumi (Fujitsu) <[email protected]>; [email protected] <[email protected]>; Euler Taveira <[email protected]>; Amit Kapila <[email protected]>; pgsql-hackers; Bharath Rupireddy <[email protected]>
Dear Melih,
Thank you for updating the patch! Followings are my comments.
01. catalogs.sgml
```
If true, the subscription will request that the publisher send data
- in binary format
```
I'm not clear here. subbinary does not directly mean that whether the worker
requests to send data or not. How about:
If true, the subscription will request that the publisher send data in binary
format, except initial data synchronization
02. create_subscription.sgml
```
+ the binary format is very data type specific, it will not allow copying
+ between different column types as opposed to text format. Note that
```
The name of formats are not specified as <literal>, whereas in previous sentence
they are. We can use format either of them.
03. parse_subscription_options()
I'm not sure the combination of "copy_format = binary" and "copy_data = false"
should be accepted or not. How do you think?
04. parse_subscription_options()
```
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("%s and %s are mutually exclusive options",
+ "binary = false", "copy_format = binary")));
+ }
```
A comment for translator seemed to be missed.
05. CreateSubscription()
```
values[Anum_pg_subscription_suborigin - 1] =
CStringGetTextDatum(opts.origin);
+ values[Anum_pg_subscription_subcopyformat - 1] = CharGetDatum(opts.copy_format);
```
I think this should be done the same ordering with FormData_pg_subscription.
Maybe after the line?
```
values[Anum_pg_subscription_subdisableonerr - 1] = BoolGetDatum(opts.disableonerr);
```
06. AlterSubscription()
If we decided not to accept "copy_format = binary" and "copy_data = false", here
should be also fixed.
07. AlterSubscription()
```
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot set %s for a subscription with %s",
+ "binary = false", "copy_format = binary")));
...
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot set %s for a subscription with %s",
+ "copy_format = binary", "binary = false")));
```
Comments for translator seemed to be missed.
08. defGetCopyFormat()
```
+ /*
+ * If no parameter value given, set it to text format.
+ */
+ if (!def->arg)
+ return LOGICALREP_COPY_AS_TEXT;
```
I think the case no parameter is given should be rejected. It should be accepted
only when the parameter has boolean data type. Note that defGetStreamingMode()
is accepted no-parameter style due to the compatibility. At first streaming is
boolean, and then "parallel" is added.
09. describeSubscriptions
```
+ /* Copy format is only supported in v16 and higher */
```
I think this comment should be atop of if statement, and it can mention about Origin too.
10. pg_subscription.h
```
+ char subcopyformat BKI_DEFAULT(LOGICALREP_COPY_AS_TEXT); /* Copy format *
```
I'm not sure whether BKI_DEFAULT() is needed or not. Other options like twophase
does not have the default value as catalog level. The default is set in
parse_subscription_options() and then the value will be set to catalog.
11. typedef struct Subscription
In catalog entry the subcopyformat is aligned just after subdisableonerr, but in
struct Subscription, copyformat is added at the end. Can we place just after disableonerr?
12. Reply
> Since binary copy relies on COPY command, we may have problems across
> different server versions in cases where COPY is not portable.
> What I understand from this [1], COPY works across server versions later
> than 7.4. This shouldn't be a problem for logical replication.
> Currently the patch also does not allow binary copy if the publisher
> version is older than 16.
If in future version the general data type is added, the copy command in binary
format will not work well, right? It is because the inferior version does not have
recv/send functions for added type. It means that there is a possibility that
replication between different versions may be failed if binary format is specified.
Therefore, I think we should accept copy_format = binary only when the major
version of servers are the same.
Note that this comments is not the request to the patch. Maybe the modification
should be done not only for copy_format but also binary, and it may be out of scope
the patch.
Best Regards,
Hayato Kuroda
FUJITSU LIMITED
^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2023-02-24 03:02 Peter Smith <[email protected]>
parent: Melih Mutlu <[email protected]>
3 siblings, 1 reply; 14+ messages in thread
From: Peter Smith @ 2023-02-24 03:02 UTC (permalink / raw)
To: Melih Mutlu <[email protected]>; +Cc: Hayato Kuroda (Fujitsu) <[email protected]>; Takamichi Osumi (Fujitsu) <[email protected]>; [email protected] <[email protected]>; Euler Taveira <[email protected]>; Amit Kapila <[email protected]>; pgsql-hackers; Bharath Rupireddy <[email protected]>
Here is my summary of this thread so far, plus some other thoughts.
(I wrote this mostly for my own internal notes/understanding, but
maybe it is a helpful summary for others so I am posting it here too).
~~
1. Initial Goal
------------------
Currently, the CREATE SUBSCRIPTION ... WITH(binary=true) mode does
data replication in binary mode, but tablesync COPY phase is still
only in text mode. IIUC Melih just wanted to unify those phases so
binary=true would mean also do the COPY phase in binary [1].
Actually, this was my very first expectation too.
2. Objections to unification
-----------------------------------
Bharath posted suggesting tying the COPY/replication parts is not a
good idea [2]. But if these are not to be unified then it requires a
new subscription option to be introduced, and currently, the thread
refers to this new option as copy_format=binary.
3. A problem: binary replication is not always binary!
----------------------------------------------------------------------
Shi-san reported [3] that under special circumstances (e.g. if the
datatype has no binary output function) the current HEAD binary=true
mode for replication has the ability to fall back to text replication.
Since the binary COPY doesn't do this, it means binary COPY could fail
in some cases where the binary=true replication will be OK.
AFAIK, this means that even if we *wanted* to unify everything with
just 'binary=true' it can't be done like that.
4. New option is needed
---------------------------------
OK, so let's assume these options must be separated (because of the
problem of #3).
~
4a. New string option called copy_format?
Currently, the patch/thread describes a new 'copy_format' string
option with values 'text' (default) and 'binary'.
Why? If there are only 2 values then IMO it would be better to have a
*boolean* option called something like binary_copy=true/false.
~
4b. Or, we could extend copy_data
Jelte suggested [4] we could extend copy_data = 'text' (aka on/true)
OR 'binary' OR 'none' (aka off/false).
That was interesting, although
- my first impression was to worry about backward compatibility issues
for existing application code. I don't know if those are easily
solved.
- AFAIK such "hybrid" boolean/enum options are kind of frowned upon so
I am not sure if a committer would be in favour of introducing another
one.
5. Combining options
------------------------------
Because options are separated, it means combinations become possible...
~~
5a. Combining option: "copy_format = binary" and "copy_data = false"
Kuroda [5] wrote such a combination should not be allowed.
I kind of disagree with that. IMO everything should be flexible as
possible. The patch might end up accidentally stopping something that
has a valid use case. Anyway, such restrictions are easy to add later.
~~
5b. Combining options: binary=true/copy_format=binary, and
binary=false/copy_format=binary become possible.
AFAIK currently the patch disallows some combinations:
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("%s and %s are mutually exclusive options",
+ "binary = false", "copy_format = binary")));
I kind of disagree with introducing such rules/restrictions. IMO all
this patch needs to do is clearly document all necessary precautions
etc. But if the user still wants to do something, we should just let
them do it. If they manage to shoot themselves in the foot, well it
was their choice after reading the docs, and it's their foot.
6. pub/sub version checking
----------------------------
There were some discussions about doing some server checking to reject
some PG combinations from being allowed to use the copy_format=binary.
Jelte suggested [5] that it is the "responsibility of the operator to
evaluate the risk".
I agree. Yes, the patch certainly needs to *document* all precautions,
but having too many restrictions might end up accidentally stopping
something useful. Anyway, this seems like #5a. I prefer KISS
Principle. More restrictions can be added later if found necessary.
7. More doubts & a thought bubble
---------------------------------
7a. What is user action for this scenario?
I am unsure about this scenario - imagine that everything is working
properly and the copy_format=binary/copy_data=true is all working
nicely for weeks for a certain pub/sub...
But if the publication was defined as FOR ALL TABLES, or as ALL TABLES
IN SCHEMA, then I think the tablesync can still crash if a user
creates a new table that suffers the same COPY binary trouble Shi-san
described earlier [3].
What is the user supposed to do when that tablesync fails?
They had no way to predict it could happen, And it will be too painful
to have to DROP and re-create the entire SUBSCRIPTION again now that
it has happened.
~~
7a. A thought bubble
I wondered (perhaps this is naive) would it be it possible to enhance
the COPY command to enhance the "binary" mode so it can be made to
fall back to text mode if it needs to in the same way that binary
replication does.
If such an enhanced COPY format mode worked, then most of the patch is redundant
- there is no need for any new option
- tablesync COPY could then *always* use this "binary-with-falback" mode
------
[1] Melih initially wanted a unified binary mode -
https://www.postgresql.org/message-id/CAGPVpCQYi9AYQSS%3DRmGgVNjz5ZEnLB8mACwd9aioVhLmbgiAMA%40mail.g...
[2] Barath doesn't like the binary/copy_format inter-dependency -
https://www.postgresql.org/message-id/CALj2ACVPt-BaLMm3Ezy1-rfUzH9qStxePcyGrHPamPESEZSBFA%40mail.gma...
[3] Shi-san found binary mode has the ability to fall back to text
sometimes - https://www.postgresql.org/message-id/OSZPR01MB6310B58F069FF8E148B247FDFDA49%40OSZPR01MB6310.jpnprd0...
[4] Jelte idea to enhance the copy_data option -
https://www.postgresql.org/message-id/CAGECzQS393xiME%2BEySLU7ceO4xOB81kPjPqNBjaeW3sLgfLhNw%40mail.g...
[5] Kuroda-san etc expecting copy_data=false/copy_format=binary
combination is not allowed -
https://www.postgresql.org/message-id/TYAPR01MB5866DDF02B3CEE59DA024CC3F5AB9%40TYAPR01MB5866.jpnprd0...
[6] Jelte says it is the operator's responsibility to use the correct
options - https://www.postgresql.org/message-id/CAGECzQSStdb%2Bx1BxzCktZd1uSjvd6eYq1wcHV3vPCykrGqxYKQ%40mail.g...
Kind Regards,
Peter Smith.
Fujitsu Australia
^ permalink raw reply [nested|flat] 14+ messages in thread
* RE: Allow logical replication to copy tables in binary format
@ 2023-02-26 22:58 Takamichi Osumi (Fujitsu) <[email protected]>
parent: Melih Mutlu <[email protected]>
3 siblings, 0 replies; 14+ messages in thread
From: Takamichi Osumi (Fujitsu) @ 2023-02-26 22:58 UTC (permalink / raw)
To: 'Melih Mutlu' <[email protected]>; Hayato Kuroda (Fujitsu) <[email protected]>; +Cc: [email protected] <[email protected]>; Euler Taveira <[email protected]>; Amit Kapila <[email protected]>; pgsql-hackers; Bharath Rupireddy <[email protected]>
On Monday, February 20, 2023 8:47 PM Melih Mutlu <[email protected]> wrote:
> Thanks for letting me know.
> Attached the fixed version of the patch.
Hi, Melih
Thanks for updating the patch. Minor comments on v9.
(1) commit message
"The patch introduces a new parameter, copy_format, to CREATE SUBSCRIPTION to allow to choose
the format used in initial table synchronization."
This patch introduces the new parameter not only to CREATE SUBSCRIPTION and ALTER SUBSCRIPTION, then this description should be more general, something like below.
"The patch introduces a new parameter, copy_format, as subscription option to
allow user to choose the format of initial table synchronization."
(2) copy_table
We don't need to check the publisher's version below.
+
+ /* If the publisher is v16 or later, specify the format to copy data. */
+ if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 160000)
+ {
+ char *format = MySubscription->copyformat == LOGICALREP_COPY_AS_BINARY ? "binary" : "text";
+ appendStringInfo(&cmd, " WITH (FORMAT %s)", format);
+ options = lappend(options, makeDefElem("format", (Node *) makeString(format), -1));
+ }
+
We don't have this kind of check for "binary" option and it seems this is user's responsibility to avoid any errors during replication. If we want to add this kind of check, then we can add checks for both "binary" and "copy_format" option together as an independent patch.
(3) subscription.sql/out
The other existing other subscription options check the invalid input for newly created option (e.g. "foo" for disable_on_error, streaming mode). So, I think we can add this type of test for this feature.
Best Regards,
Takamichi Osumi
^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2023-02-27 10:16 Bharath Rupireddy <[email protected]>
parent: Peter Smith <[email protected]>
0 siblings, 0 replies; 14+ messages in thread
From: Bharath Rupireddy @ 2023-02-27 10:16 UTC (permalink / raw)
To: Peter Smith <[email protected]>; +Cc: Melih Mutlu <[email protected]>; Hayato Kuroda (Fujitsu) <[email protected]>; Takamichi Osumi (Fujitsu) <[email protected]>; [email protected] <[email protected]>; Euler Taveira <[email protected]>; Amit Kapila <[email protected]>; pgsql-hackers
On Fri, Feb 24, 2023 at 8:32 AM Peter Smith <[email protected]> wrote:
>
> Here is my summary of this thread so far, plus some other thoughts.
Thanks.
> 1. Initial Goal
> ------------------
>
> Currently, the CREATE SUBSCRIPTION ... WITH(binary=true) mode does
> data replication in binary mode, but tablesync COPY phase is still
> only in text mode. IIUC Melih just wanted to unify those phases so
> binary=true would mean also do the COPY phase in binary [1].
>
> Actually, this was my very first expectation too.
>
> 2. Objections to unification
> -----------------------------------
>
> Bharath posted suggesting tying the COPY/replication parts is not a
> good idea [2]. But if these are not to be unified then it requires a
> new subscription option to be introduced, and currently, the thread
> refers to this new option as copy_format=binary.
Looking closely at the existing binary=true option and COPY's binary
protocol, essentially they depend on the data type's binary send and
receive functions.
> 3. A problem: binary replication is not always binary!
> ----------------------------------------------------------------------
>
> Shi-san reported [3] that under special circumstances (e.g. if the
> datatype has no binary output function) the current HEAD binary=true
> mode for replication has the ability to fall back to text replication.
> Since the binary COPY doesn't do this, it means binary COPY could fail
> in some cases where the binary=true replication will be OK.
Right. Apply workers can fallback to text mode transparently, whereas
with binary COPY it's a bit difficult to do so.
> AFAIK, this means that even if we *wanted* to unify everything with
> just 'binary=true' it can't be done like that.
Hm, it looks like that.
> 4. New option is needed
> ---------------------------------
>
> OK, so let's assume these options must be separated (because of the
> problem of #3).
>
> ~
>
> 4a. New string option called copy_format?
>
> Currently, the patch/thread describes a new 'copy_format' string
> option with values 'text' (default) and 'binary'.
>
> Why? If there are only 2 values then IMO it would be better to have a
> *boolean* option called something like binary_copy=true/false.
>
> ~
>
> 4b. Or, we could extend copy_data
>
> Jelte suggested [4] we could extend copy_data = 'text' (aka on/true)
> OR 'binary' OR 'none' (aka off/false).
How about copy_binary = {true/false}? So, the options for the user are:
copy_binary - defaults to false and when true, the subscriber requests
publisher to send pre-existing table's data in binary format (as
opposed to text) during data synchronization/table copy phase. It is
recommended to enable this option only when 1) the column data types
have appropriate binary send/receive functions, 2) not replicating
between different major versions or different platforms, 3) both
publisher and subscriber tables have the exact same column types (not
when replicating from smallint to int or numeric to int8 and so on), 4)
both publisher and subscriber supports COPY with binary option,
otherwise the table copy can fail.
binary - defaults to false and when true, the subscriber requests
publisher to send table's data in binary format (as opposed to text)
during normal replication phase. <<existing description from the docs
continues>>
Note that with this we made users responsible to choose copy_binary
rather than we being smart.
> AFAIK currently the patch disallows some combinations:
>
> + ereport(ERROR,
> + (errcode(ERRCODE_SYNTAX_ERROR),
> + errmsg("%s and %s are mutually exclusive options",
> + "binary = false", "copy_format = binary")));
>
> 6. pub/sub version checking
> ----------------------------
>
> There were some discussions about doing some server checking to reject
> some PG combinations from being allowed to use the copy_format=binary.
IMHO, these restrictions make the feature more complicated to use and
be removed and the options be made simple to use (usability and
simplicity clearly wins the race). If there's any kind of feedback
from the users/developers, we can always come back and improve.
> But if the publication was defined as FOR ALL TABLES, or as ALL TABLES
> IN SCHEMA, then I think the tablesync can still crash if a user
> creates a new table that suffers the same COPY binary trouble Shi-san
> described earlier [3].
>
> What is the user supposed to do when that tablesync fails?
>
> They had no way to predict it could happen, And it will be too painful
> to have to DROP and re-create the entire SUBSCRIPTION again now that
> it has happened.
Can't ALTER SUBSCRIPTION .... SET copy_format = 'text'; and ALTER
SUBSCRIPTION ... REFRESH PUBLICATION WITH (copy_data = true); work
here instead of drop and recreate subscription?
> 7a. A thought bubble
>
> I wondered (perhaps this is naive) would it be it possible to enhance
> the COPY command to enhance the "binary" mode so it can be made to
> fall back to text mode if it needs to in the same way that binary
> replication does.
>
> If such an enhanced COPY format mode worked, then most of the patch is redundant
> - there is no need for any new option
> - tablesync COPY could then *always* use this "binary-with-falback" mode
I don't think that's a great idea. COPY is a user-facing SQL command
and any errors (because of data type not having binary send/receive
functions) better be reported to the users. Again, such an option
might complicate both COPY code and usability.
--
Bharath Rupireddy
PostgreSQL Contributors Team
RDS Open Source Databases
Amazon Web Services: https://aws.amazon.com
^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2023-02-27 13:24 Bharath Rupireddy <[email protected]>
parent: Bharath Rupireddy <[email protected]>
0 siblings, 1 reply; 14+ messages in thread
From: Bharath Rupireddy @ 2023-02-27 13:24 UTC (permalink / raw)
To: Melih Mutlu <[email protected]>; +Cc: Hayato Kuroda (Fujitsu) <[email protected]>; Takamichi Osumi (Fujitsu) <[email protected]>; [email protected] <[email protected]>; Euler Taveira <[email protected]>; Amit Kapila <[email protected]>; pgsql-hackers
On Tue, Feb 21, 2023 at 7:18 PM Bharath Rupireddy
<[email protected]> wrote:
>
> On Mon, Feb 20, 2023 at 5:17 PM Melih Mutlu <[email protected]> wrote:
> >
> > Thanks for letting me know.
> > Attached the fixed version of the patch.
>
> Thanks. I have few comments on v9 patch:
>
> 1.
> + /* Do not allow binary = false with copy_format = binary */
> + if (!opts.binary &&
> + sub->copyformat == LOGICALREP_COPY_AS_BINARY &&
> + !IsSet(opts.specified_opts, SUBOPT_COPY_FORMAT))
> + ereport(ERROR,
> +
> (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
> + errmsg("cannot set %s for a
> subscription with %s",
> + "binary = false",
> "copy_format = binary")));
>
> I don't understand why we'd need to tie an option (binary) that deals
> with data types at column-level with another option (copy_format) that
> requests the entire table data to be in binary. This'd essentially
> make one to set binary = true to use copy_format = binary, no? IMHO,
> this inter-dependency is not good for better usability.
>
> 2. Why can't the tests that this patch adds be simple? Why would it
> need to change the existing tests at all? I'm thinking to create a new
> 00X_binary_copy_format.pl or such and setting up logical replication
> with copy_format = binary and letting table sync worker request
> publisher in binary format - you can verify this via publisher server
> logs - look for COPY with BINARY option. If required, have the table
> with different data types. This greatly reduces the patch's footprint.
I've done performance testing with the v9 patch.
I can constantly observe 1.34X improvement or 25% improvement in table
sync/copy performance with the patch:
HEAD binary = false
Time: 214398.637 ms (03:34.399)
PATCHED binary = true, copy_format = binary:
Time: 160509.262 ms (02:40.509)
There's a clear reduction (5.68% to 4.49%) in the CPU cycles spent in
copy_table with the patch:
perf report HEAD:
- 5.68% 0.00% postgres postgres [.] copy_table
- copy_table
- 5.67% CopyFrom
- 4.26% NextCopyFrom
- 2.16% NextCopyFromRawFields
- 1.55% CopyReadLine
- 1.52% CopyReadLineText
- 0.76% CopyLoadInputBuf
0.50% CopyConvertBuf
0.60% CopyReadAttributesText
- 1.93% InputFunctionCall
0.69% timestamptz_in
0.53% byteain
- 0.73% CopyMultiInsertInfoFlush
- 0.73% CopyMultiInsertBufferFlush
- 0.66% table_multi_insert
0.65% heap_multi_insert
perf report PATCHED:
- 4.49% 0.00% postgres postgres [.] copy_table
- copy_table
- 4.48% CopyFrom
- 2.36% NextCopyFrom
- 1.81% CopyReadBinaryAttribute
1.47% ReceiveFunctionCall
- 1.21% CopyMultiInsertInfoFlush
- 1.21% CopyMultiInsertBufferFlush
- 1.11% table_multi_insert
- 1.09% heap_multi_insert
- 0.71% RelationGetBufferForTuple
- 0.63% ReadBufferBI
- 0.62% ReadBufferExtended
0.62% ReadBuffer_common
I've tried to choose the table columns such that the binary send/recv
vs non-binary/plain/text copy has some benefit. The table has 100mn
rows, and is of 11GB size. I've measured the benefit using Melih's
helper function wait_for_rep(). Note that I had to compile source code
with -ggdb3 -O0 for perf report, otherwise with -O3 for performance
numbers:
wal_level = 'logical'
shared_buffers = '8GB'
wal_buffers = '1GB'
max_wal_size = '32GB'
create table foo(i int, n numeric, v varchar, b bytea, t timestamp
with time zone default current_timestamp);
insert into foo select i, i+1, md5(i::text), md5(i::text)::bytea from
generate_series(1, 100000000) i;
CREATE OR REPLACE PROCEDURE wait_for_rep()
LANGUAGE plpgsql
AS $$
BEGIN
WHILE (SELECT count(*) != 0 FROM pg_subscription_rel WHERE
srsubstate <> 'r') LOOP COMMIT;
END LOOP;
END;
$$;
--
Bharath Rupireddy
PostgreSQL Contributors Team
RDS Open Source Databases
Amazon Web Services: https://aws.amazon.com
^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2023-02-27 19:52 Melih Mutlu <[email protected]>
parent: Bharath Rupireddy <[email protected]>
0 siblings, 2 replies; 14+ messages in thread
From: Melih Mutlu @ 2023-02-27 19:52 UTC (permalink / raw)
To: Bharath Rupireddy <[email protected]>; Takamichi Osumi (Fujitsu) <[email protected]>; Jelte Fennema <[email protected]>; Amit Kapila <[email protected]>; +Cc: pgsql-hackers; Hayato Kuroda (Fujitsu) <[email protected]>; [email protected] <[email protected]>; Euler Taveira <[email protected]>
Hi,
Thanks for all of your reviews!
So, I made some changes in the v10 according to your comments.
1- copy_format is changed to a boolean parameter copy_binary.
It sounds easier to use a boolean to enable/disable binary copy. Default
value is false, so nothing changes in the current behaviour if copy_binary
is not specified.
It's still persisted into the catalog. This is needed since its value will
be needed by tablesync workers later. And tablesync workers fetch
subscription configurations from the catalog.
In the copy_data case, it is not directly stored anywhere but it affects
the state of the table which is stored in the catalog. So, I guess this is
the convenient way if we decide to go with a new parameter.
2- copy_binary is not tied to another parameter
The patch does not disallow any combination of copy_binary with binary and
copy_data options. I tried to explain what binary copy can and cannot do in
the docs. Rest is up to the user now.
3- Removed version check for copy_binary
HEAD already does not have any check for binary option. Making binary copy
work only if publisher and subscriber are the same major version can be too
restricted.
4- Added separate test file
Although I believe 002_types.pl and 014_binary.pl can be improved more for
binary enabled subscription cases, this patch might not be the best place
to make those changes.
032_binary_copy.pl now has the tests for binary copy option. There are also
some regression tests in subscription.sql.
Finally, some other small fixes are done according to the reviews.
Also, thanks Bharath for performance testing [1]. It can be seen that the
patch has some benefits.
I appreciate further thoughts/reviews.
[1]
https://www.postgresql.org/message-id/CALj2ACUfE08ZNjKK-nK9JiwGhwUMRLM%2BqRhNKTVM9HipFk7Fow%40mail.g...
Thanks,
--
Melih Mutlu
Microsoft
Attachments:
[application/octet-stream] v10-0001-Allow-logical-replication-to-copy-table-in-binar.patch (65.1K, ../../CAGPVpCTAKDraT9Y8OW3u0b+sKswDcbaruNN+RN+fVE0LAYP4qg@mail.gmail.com/3-v10-0001-Allow-logical-replication-to-copy-table-in-binar.patch)
download | inline diff:
From 8fe92c301d84ed6e34cfe0e397ef14d929e07053 Mon Sep 17 00:00:00 2001
From: Melih Mutlu <[email protected]>
Date: Mon, 8 Aug 2022 14:14:44 +0300
Subject: [PATCH v10] Allow logical replication to copy table in binary
If binary option is enabled in a subscription and copy_format is set to binary, then copy tables in
binary format during table synchronization.
The patch introduces a new parameter, copy_binary, as subscription option to
allow users to choose the format of initial table synchronization
Without this patch, table are copied in text format even if the
subscription is created with binary option. When binary format is
beneficial to use, allowing the subscription to copy tables in binary in
table sync phase may reduce the time spent on copy depending on column types.
Discussion: https://postgr.es/m/CAGPVpCQvAziCLknEnygY0v1-KBtg%2BOm-9JHJYZOnNPKFJPompw%40mail.gmail.com
---
doc/src/sgml/catalogs.sgml | 9 +
doc/src/sgml/logical-replication.sgml | 9 +-
doc/src/sgml/ref/alter_subscription.sgml | 20 +-
doc/src/sgml/ref/create_subscription.sgml | 16 ++
src/backend/catalog/pg_subscription.c | 1 +
src/backend/catalog/system_views.sql | 3 +-
src/backend/commands/subscriptioncmds.c | 57 ++++-
src/backend/replication/logical/tablesync.c | 10 +-
src/bin/pg_dump/pg_dump.c | 17 +-
src/bin/pg_dump/pg_dump.h | 1 +
src/bin/psql/describe.c | 11 +-
src/bin/psql/tab-complete.c | 14 +-
src/include/catalog/pg_subscription.h | 2 +
src/test/regress/expected/subscription.out | 178 +++++++++-------
src/test/regress/sql/subscription.sql | 18 ++
src/test/subscription/meson.build | 1 +
src/test/subscription/t/032_binary_copy.pl | 223 ++++++++++++++++++++
17 files changed, 493 insertions(+), 97 deletions(-)
create mode 100644 src/test/subscription/t/032_binary_copy.pl
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index c1e4048054..90f886b916 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -8002,6 +8002,15 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
origin.
</para></entry>
</row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>subcopybinary</structfield> <type>bool</type>
+ </para>
+ <para>
+ If true, initial data synchronization will be performed in binary format
+ </para></entry>
+ </row>
</tbody>
</tgroup>
</table>
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 1bd5660c87..83caf40a05 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -241,10 +241,11 @@
types of the columns do not need to match, as long as the text
representation of the data can be converted to the target type. For
example, you can replicate from a column of type <type>integer</type> to a
- 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.
+ column of type <type>bigint</type>. However, replication in binary format is
+ type specific and does not allow to replicate data between different types
+ according to its restrictions. 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.
</para>
<sect2 id="logical-replication-subscription-slot">
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 964fcbb8ff..b962ccd986 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -179,6 +179,22 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
</para>
</listitem>
</varlistentry>
+
+ <varlistentry>
+ <term><literal>copy_binary</literal> (<type>boolean</type>)</term>
+ <listitem>
+ <para>
+ Specifies whether pre-existing data on the publisher will be copied
+ to the subscriber in binary format. The default is <literal>false</literal>.
+ Binary format is very data type specific, it will not allow copying
+ between different column types as opposed to text format. Note that
+ if this option is enabled, all data types which will be copied during
+ the initial synchronization should have binary send and receive functions.
+ If this option is disabled, data format for the initial synchronization
+ will be text.
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist></para>
</listitem>
</varlistentry>
@@ -213,8 +229,8 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
are <literal>slot_name</literal>,
<literal>synchronous_commit</literal>,
<literal>binary</literal>, <literal>streaming</literal>,
- <literal>disable_on_error</literal>, and
- <literal>origin</literal>.
+ <literal>disable_on_error</literal>, <literal>origin</literal>
+ and <literal>copy_binary</literal>.
</para>
</listitem>
</varlistentry>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 51c45f17c7..a2ca8adffc 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -349,6 +349,22 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
</para>
</listitem>
</varlistentry>
+
+ <varlistentry>
+ <term><literal>copy_binary</literal> (<type>boolean</type>)</term>
+ <listitem>
+ <para>
+ Specifies whether pre-existing data on the publisher will be copied
+ to the subscriber in binary format. The default is <literal>false</literal>.
+ Binary format is very data type specific, it will not allow copying
+ between different column types as opposed to text format. Note that
+ if this option is enabled, all data types which will be copied during
+ the initial synchronization should have binary send and receive functions.
+ If this option is disabled, data format for the initial synchronization
+ will be text.
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist></para>
</listitem>
diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index a56ae311c3..ffd3324bc1 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -71,6 +71,7 @@ GetSubscription(Oid subid, bool missing_ok)
sub->stream = subform->substream;
sub->twophasestate = subform->subtwophasestate;
sub->disableonerr = subform->subdisableonerr;
+ sub->copybinary = subform->subcopybinary;
/* Get conninfo */
datum = SysCacheGetAttr(SUBSCRIPTIONOID,
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 34ca0e739f..2521606bbc 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1316,7 +1316,8 @@ REVOKE ALL ON pg_replication_origin_status FROM public;
REVOKE ALL ON pg_subscription FROM public;
GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled,
subbinary, substream, subtwophasestate, subdisableonerr,
- subslotname, subsynccommit, subpublications, suborigin)
+ subslotname, subsynccommit, subpublications, suborigin,
+ subcopybinary)
ON pg_subscription TO public;
CREATE VIEW pg_stat_subscription_stats AS
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 464db6d247..7c3d67ad73 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -66,6 +66,7 @@
#define SUBOPT_DISABLE_ON_ERR 0x00000400
#define SUBOPT_LSN 0x00000800
#define SUBOPT_ORIGIN 0x00001000
+#define SUBOPT_COPY_BINARY 0x00002000
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
@@ -90,6 +91,7 @@ typedef struct SubOpts
bool disableonerr;
char *origin;
XLogRecPtr lsn;
+ bool copy_binary;
} SubOpts;
static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
@@ -146,6 +148,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->disableonerr = false;
if (IsSet(supported_opts, SUBOPT_ORIGIN))
opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY);
+ if (IsSet(supported_opts, SUBOPT_COPY_BINARY))
+ opts->copy_binary = false;
/* Parse options */
foreach(lc, stmt_options)
@@ -324,6 +328,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->specified_opts |= SUBOPT_LSN;
opts->lsn = lsn;
}
+ else if (IsSet(supported_opts, SUBOPT_COPY_BINARY) &&
+ strcmp(defel->defname, "copy_binary") == 0)
+ {
+ if (IsSet(opts->specified_opts, SUBOPT_COPY_BINARY))
+ errorConflictingDefElem(defel, pstate);
+
+ opts->specified_opts |= SUBOPT_COPY_BINARY;
+ opts->copy_binary = defGetBoolean(defel);
+ }
else
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
@@ -560,7 +573,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
SUBOPT_SLOT_NAME | SUBOPT_COPY_DATA |
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
- SUBOPT_DISABLE_ON_ERR | SUBOPT_ORIGIN);
+ SUBOPT_DISABLE_ON_ERR | SUBOPT_ORIGIN |
+ SUBOPT_COPY_BINARY);
parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
/*
@@ -636,6 +650,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
LOGICALREP_TWOPHASE_STATE_PENDING :
LOGICALREP_TWOPHASE_STATE_DISABLED);
values[Anum_pg_subscription_subdisableonerr - 1] = BoolGetDatum(opts.disableonerr);
+ values[Anum_pg_subscription_subcopybinary - 1] = BoolGetDatum(opts.copy_binary);
values[Anum_pg_subscription_subconninfo - 1] =
CStringGetTextDatum(conninfo);
if (opts.slot_name)
@@ -1054,7 +1069,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
supported_opts = (SUBOPT_SLOT_NAME |
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
SUBOPT_STREAMING | SUBOPT_DISABLE_ON_ERR |
- SUBOPT_ORIGIN);
+ SUBOPT_ORIGIN | SUBOPT_COPY_BINARY);
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -1118,6 +1133,13 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
replaces[Anum_pg_subscription_suborigin - 1] = true;
}
+ if (IsSet(opts.specified_opts, SUBOPT_COPY_BINARY))
+ {
+ values[Anum_pg_subscription_subcopybinary - 1] =
+ BoolGetDatum(opts.copy_binary);
+ replaces[Anum_pg_subscription_subcopybinary - 1] = true;
+ }
+
update_tuple = true;
break;
}
@@ -1158,7 +1180,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
case ALTER_SUBSCRIPTION_SET_PUBLICATION:
{
- supported_opts = SUBOPT_COPY_DATA | SUBOPT_REFRESH;
+ supported_opts = (SUBOPT_COPY_DATA | SUBOPT_REFRESH |
+ SUBOPT_COPY_BINARY);
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -1166,6 +1189,13 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
publicationListToArray(stmt->publication);
replaces[Anum_pg_subscription_subpublications - 1] = true;
+ if (IsSet(opts.specified_opts, SUBOPT_COPY_BINARY))
+ {
+ values[Anum_pg_subscription_subcopybinary - 1] =
+ BoolGetDatum(opts.copy_binary);
+ replaces[Anum_pg_subscription_subcopybinary - 1] = true;
+ }
+
update_tuple = true;
/* Refresh if user asked us to. */
@@ -1205,7 +1235,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
List *publist;
bool isadd = stmt->kind == ALTER_SUBSCRIPTION_ADD_PUBLICATION;
- supported_opts = SUBOPT_REFRESH | SUBOPT_COPY_DATA;
+ supported_opts = (SUBOPT_REFRESH | SUBOPT_COPY_DATA |
+ SUBOPT_COPY_BINARY);
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -1214,6 +1245,13 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
publicationListToArray(publist);
replaces[Anum_pg_subscription_subpublications - 1] = true;
+ if (IsSet(opts.specified_opts, SUBOPT_COPY_BINARY))
+ {
+ values[Anum_pg_subscription_subcopybinary - 1] =
+ BoolGetDatum(opts.copy_binary);
+ replaces[Anum_pg_subscription_subcopybinary - 1] = true;
+ }
+
update_tuple = true;
/* Refresh if user asked us to. */
@@ -1265,8 +1303,17 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("ALTER SUBSCRIPTION ... REFRESH is not allowed for disabled subscriptions")));
+ supported_opts = SUBOPT_COPY_DATA | SUBOPT_COPY_BINARY;
parse_subscription_options(pstate, stmt->options,
- SUBOPT_COPY_DATA, &opts);
+ supported_opts, &opts);
+
+ if (IsSet(opts.specified_opts, SUBOPT_COPY_BINARY))
+ {
+ values[Anum_pg_subscription_subcopybinary - 1] =
+ BoolGetDatum(opts.copy_binary);
+ replaces[Anum_pg_subscription_subcopybinary - 1] = true;
+ }
+ update_tuple = true;
/*
* The subscription option "two_phase" requires that
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 07eea504ba..c14d32fe07 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -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"
@@ -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)),
@@ -1168,6 +1170,12 @@ copy_table(Relation rel)
appendStringInfoString(&cmd, ") TO STDOUT");
}
+
+ appendStringInfo(&cmd, " WITH (FORMAT %s)", MySubscription->copybinary ? "binary" : "text");
+ options = lappend(options, makeDefElem("format",
+ (Node *) makeString(MySubscription->copybinary ? "binary" : "text"),
+ -1));
+
res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 0, NULL);
pfree(cmd.data);
if (res->status != WALRCV_OK_COPY_OUT)
@@ -1184,7 +1192,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);
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 24ba936332..1a863464c7 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4493,6 +4493,7 @@ getSubscriptions(Archive *fout)
int i_subsynccommit;
int i_subpublications;
int i_subbinary;
+ int i_subcopybinary;
int i,
ntups;
@@ -4545,9 +4546,15 @@ getSubscriptions(Archive *fout)
LOGICALREP_TWOPHASE_STATE_DISABLED);
if (fout->remoteVersion >= 160000)
- appendPQExpBufferStr(query, " s.suborigin\n");
+ {
+ appendPQExpBufferStr(query, " s.suborigin,\n"
+ " s.subcopybinary\n");
+ }
else
- appendPQExpBuffer(query, " '%s' AS suborigin\n", LOGICALREP_ORIGIN_ANY);
+ {
+ appendPQExpBuffer(query, " '%s' AS suborigin,\n", LOGICALREP_ORIGIN_ANY);
+ appendPQExpBuffer(query, " false AS subcopybinary\n");
+ }
appendPQExpBufferStr(query,
"FROM pg_subscription s\n"
@@ -4575,6 +4582,7 @@ getSubscriptions(Archive *fout)
i_subtwophasestate = PQfnumber(res, "subtwophasestate");
i_subdisableonerr = PQfnumber(res, "subdisableonerr");
i_suborigin = PQfnumber(res, "suborigin");
+ i_subcopybinary = PQfnumber(res, "subcopybinary");
subinfo = pg_malloc(ntups * sizeof(SubscriptionInfo));
@@ -4605,6 +4613,8 @@ getSubscriptions(Archive *fout)
subinfo[i].subdisableonerr =
pg_strdup(PQgetvalue(res, i, i_subdisableonerr));
subinfo[i].suborigin = pg_strdup(PQgetvalue(res, i, i_suborigin));
+ subinfo[i].subcopybinary =
+ pg_strdup(PQgetvalue(res, i, i_subcopybinary));
/* Decide whether we want to dump it */
selectDumpableObject(&(subinfo[i].dobj), fout);
@@ -4686,6 +4696,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
if (strcmp(subinfo->subsynccommit, "off") != 0)
appendPQExpBuffer(query, ", synchronous_commit = %s", fmtId(subinfo->subsynccommit));
+ if (strcmp(subinfo->subcopybinary, "t") == 0)
+ appendPQExpBuffer(query, ", copy_binary = true");
+
appendPQExpBufferStr(query, ");\n");
if (subinfo->dobj.dump & DUMP_COMPONENT_DEFINITION)
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index cdca0b993d..dc1227e494 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -661,6 +661,7 @@ typedef struct _SubscriptionInfo
char *suborigin;
char *subsynccommit;
char *subpublications;
+ char *subcopybinary;
} SubscriptionInfo;
/*
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index c8a0bb7b3a..edace80acb 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -6472,7 +6472,7 @@ describeSubscriptions(const char *pattern, bool verbose)
PGresult *res;
printQueryOpt myopt = pset.popt;
static const bool translate_columns[] = {false, false, false, false,
- false, false, false, false, false, false, false, false};
+ false, false, false, false, false, false, false, false, false};
if (pset.sversion < 100000)
{
@@ -6527,10 +6527,15 @@ describeSubscriptions(const char *pattern, bool verbose)
gettext_noop("Two-phase commit"),
gettext_noop("Disable on error"));
+ /* Copy binary and origin are only supported in v16 and higher */
if (pset.sversion >= 160000)
+ {
appendPQExpBuffer(&buf,
- ", suborigin AS \"%s\"\n",
- gettext_noop("Origin"));
+ ", suborigin AS \"%s\"\n"
+ ", subcopybinary AS \"%s\"\n",
+ gettext_noop("Origin"),
+ gettext_noop("Copy binary"));
+ }
appendPQExpBuffer(&buf,
", subsynccommit AS \"%s\"\n"
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 5e1882eaea..9635eb2740 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1919,14 +1919,14 @@ psql_completion(const char *text, int start, int end)
/* ALTER SUBSCRIPTION <name> REFRESH PUBLICATION WITH ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) &&
TailMatches("REFRESH", "PUBLICATION", "WITH", "("))
- COMPLETE_WITH("copy_data");
+ COMPLETE_WITH("copy_binary", "copy_data");
/* ALTER SUBSCRIPTION <name> SET */
else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, "SET"))
COMPLETE_WITH("(", "PUBLICATION");
/* ALTER SUBSCRIPTION <name> SET ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "("))
- COMPLETE_WITH("binary", "disable_on_error", "origin", "slot_name",
- "streaming", "synchronous_commit");
+ COMPLETE_WITH("binary", "copy_binary", "disable_on_error", "origin",
+ "slot_name", "streaming", "synchronous_commit");
/* ALTER SUBSCRIPTION <name> SKIP ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SKIP", "("))
COMPLETE_WITH("lsn");
@@ -1942,7 +1942,7 @@ psql_completion(const char *text, int start, int end)
/* ALTER SUBSCRIPTION <name> ADD|DROP|SET PUBLICATION <name> WITH ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) &&
TailMatches("ADD|DROP|SET", "PUBLICATION", MatchAny, "WITH", "("))
- COMPLETE_WITH("copy_data", "refresh");
+ COMPLETE_WITH("copy_binary", "copy_data", "refresh");
/* ALTER SCHEMA <name> */
else if (Matches("ALTER", "SCHEMA", MatchAny))
@@ -3267,9 +3267,9 @@ psql_completion(const char *text, int start, int end)
COMPLETE_WITH("WITH (");
/* Complete "CREATE SUBSCRIPTION <name> ... WITH ( <opt>" */
else if (HeadMatches("CREATE", "SUBSCRIPTION") && TailMatches("WITH", "("))
- COMPLETE_WITH("binary", "connect", "copy_data", "create_slot",
- "disable_on_error", "enabled", "origin", "slot_name",
- "streaming", "synchronous_commit", "two_phase");
+ COMPLETE_WITH("binary", "connect", "copy_binary", "copy_data",
+ "create_slot", "disable_on_error", "enabled", "origin",
+ "slot_name", "streaming", "synchronous_commit", "two_phase");
/* CREATE TRIGGER --- is allowed inside CREATE SCHEMA, so use TailMatches */
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index b0f2a1705d..a93a3f3dab 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -88,6 +88,7 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
bool subdisableonerr; /* True if a worker error should cause the
* subscription to be disabled */
+ bool subcopybinary; /* True if initial sync will be performed in binary*/
#ifdef CATALOG_VARLEN /* variable-length fields start here */
/* Connection string to the publisher */
text subconninfo BKI_FORCE_NOT_NULL;
@@ -131,6 +132,7 @@ typedef struct Subscription
bool disableonerr; /* Indicates if the subscription should be
* automatically disabled if a worker error
* occurs */
+ bool copybinary; /* Indicates if initial sync will be in binary */
char *conninfo; /* Connection string to the publisher */
char *slotname; /* Name of the replication slot */
char *synccommit; /* Synchronous commit setting for worker */
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 3f99b14394..700bd108b0 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -114,18 +114,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | none | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | none | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub4 SET (origin = any);
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub3;
@@ -143,10 +143,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@@ -163,10 +163,10 @@ ERROR: unrecognized subscription parameter: "create_slot"
-- ok
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345');
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | off | dbname=regress_doesnotexist2 | 0/12345
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | f | off | dbname=regress_doesnotexist2 | 0/12345
(1 row)
-- ok - with lsn = NONE
@@ -175,10 +175,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE);
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0');
ERROR: invalid WAL location (LSN): 0/0
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | off | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | f | off | dbname=regress_doesnotexist2 | 0/0
(1 row)
BEGIN;
@@ -210,10 +210,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar);
ERROR: invalid value for parameter "synchronous_commit": "foobar"
HINT: Available values: local, remote_write, remote_apply, on, off.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+--------------------+------------------------------+----------
- regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | local | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+------------------------------+----------
+ regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | f | local | dbname=regress_doesnotexist2 | 0/0
(1 row)
-- rename back to keep the rest simple
@@ -247,19 +247,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (binary = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -271,27 +271,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication already exists
@@ -306,10 +306,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr
ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false);
ERROR: publication "testpub1" is already in subscription "regress_testsub"
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication used more than once
@@ -324,10 +324,10 @@ ERROR: publication "testpub3" is not in subscription "regress_testsub"
-- ok - delete publications
ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -363,10 +363,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
--fail - alter of two_phase option not supported.
@@ -375,10 +375,10 @@ ERROR: unrecognized subscription parameter: "two_phase"
-- but can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -388,10 +388,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -404,18 +404,52 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | t | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | t | any | f | off | dbname=regress_doesnotexist | 0/0
+(1 row)
+
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+-- fail - copy_binary must be boolean
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_binary = foo);
+ERROR: copy_binary requires a Boolean value
+-- now it works - create a subscription with copy_binary = true
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_binary = true);
+WARNING: subscription was created, but is not connected
+HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
+\dRs+
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | off | dbname=regress_doesnotexist | 0/0
+(1 row)
+
+-- alter copy_binary to false
+ALTER SUBSCRIPTION regress_testsub SET (copy_binary = false);
+\dRs+
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
+(1 row)
+
+-- add publication with copy_binary = true
+ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1 WITH (refresh = false, copy_binary = true);
+\dRs+
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+--------------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub,testpub1} | f | off | d | f | any | t | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index 7281f5fee2..f6b08cbb08 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -286,6 +286,24 @@ ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
DROP SUBSCRIPTION regress_testsub;
+-- fail - copy_binary must be boolean
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_binary = foo);
+
+-- now it works - create a subscription with copy_binary = true
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_binary = true);
+\dRs+
+
+-- alter copy_binary to false
+ALTER SUBSCRIPTION regress_testsub SET (copy_binary = false);
+\dRs+
+
+-- add publication with copy_binary = true
+ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1 WITH (refresh = false, copy_binary = true);
+\dRs+
+
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+
RESET SESSION AUTHORIZATION;
DROP ROLE regress_subscription_user;
DROP ROLE regress_subscription_user2;
diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build
index 3db0fdfd96..f9ab6eb7e1 100644
--- a/src/test/subscription/meson.build
+++ b/src/test/subscription/meson.build
@@ -38,6 +38,7 @@ tests += {
't/029_on_error.pl',
't/030_origin.pl',
't/031_column_list.pl',
+ 't/032_binary_copy.pl',
't/100_bugs.pl',
],
},
diff --git a/src/test/subscription/t/032_binary_copy.pl b/src/test/subscription/t/032_binary_copy.pl
new file mode 100644
index 0000000000..bcad66e5ea
--- /dev/null
+++ b/src/test/subscription/t/032_binary_copy.pl
@@ -0,0 +1,223 @@
+
+# Copyright (c) 2023, PostgreSQL Global Development Group
+
+# Basic logical replication test
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# Initialize publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->start;
+
+# Create subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init(allows_streaming => 'logical');
+$node_subscriber->start;
+
+# Create tables on both sides of the replication
+my $ddl = qq(
+ CREATE TABLE public.test_numerical (
+ a INTEGER PRIMARY KEY,
+ b NUMERIC,
+ c FLOAT,
+ d BIGINT
+ );
+ CREATE TABLE public.test_arrays (
+ a INTEGER[] PRIMARY KEY,
+ b NUMERIC[],
+ c TEXT[]
+ );
+ CREATE TABLE public.test_range_array (
+ a INTEGER PRIMARY KEY,
+ b TSTZRANGE,
+ c int8range[]
+ );
+ CREATE TYPE public.test_comp_basic_t AS (a FLOAT, b TEXT, c INTEGER);
+ CREATE TABLE public.test_one_comp (
+ a INTEGER PRIMARY KEY,
+ b public.test_comp_basic_t
+ ););
+
+$node_publisher->safe_psql('postgres', $ddl);
+$node_subscriber->safe_psql('postgres', $ddl);
+
+# Publish all tables
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tpub FOR ALL TABLES");
+
+# Insert some content and 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"}');
+ INSERT INTO test_range_array (a, b, c) VALUES
+ (1, tstzrange('Mon Aug 04 00:00:00 2014 CEST'::timestamptz, 'infinity'), '{"[1,2]", "[10,20]"}'),
+ (2, tstzrange('Sat Aug 02 00:00:00 2014 CEST'::timestamptz, 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[2,3]", "[20,30]"}');
+ INSERT INTO test_one_comp (a, b) VALUES
+ (1, ROW(1.0, 'a', 1)),
+ (2, ROW(2.0, 'b', 2));
+ ));
+
+# Create the subscription with copy_binary = true
+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, copy_binary = true)");
+
+# Ensure nodes are in sync with each other
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tsub');
+
+my $sync_check = qq(
+ SET timezone = '+2';
+ SELECT a, b, c, d FROM test_numerical ORDER BY a;
+ SELECT a, b, c FROM test_arrays ORDER BY a;
+ SELECT a, b FROM test_one_comp ORDER BY a;
+ SELECT a, b, c FROM test_range_array ORDER BY a;
+);
+
+# Check the synced data on subscribers
+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}
+1|(1,a,1)
+2|(2,b,2)
+1|["2014-08-04 00:00:00+02",infinity)|{"[1,3)","[10,21)"}
+2|["2014-08-02 00:00:00+02","2014-08-04 00:00:00+02")|{"[2,4)","[20,31)"}', 'check synced data on subscriber');
+
+# 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');
+ ));
+
+# 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/);
+
+# 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');
+
+$sync_check = qq(
+ SET timezone = '+2';
+ SELECT a, b, c, d FROM test_numerical ORDER BY a;
+ SELECT a, b, c FROM test_arrays ORDER BY a;
+ SELECT a, b FROM test_one_comp ORDER BY a;
+ SELECT a, b, c FROM test_range_array ORDER BY a;
+ SELECT a FROM test_myvarchar;
+);
+
+# Check the synced data on subscriber
+$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}
+1|(1,a,1)
+2|(2,b,2)
+1|["2014-08-04 00:00:00+02",infinity)|{"[1,3)","[10,21)"}
+2|["2014-08-02 00:00:00+02","2014-08-04 00:00:00+02")|{"[2,4)","[20,31)"}
+a', 'check synced data on subscriber with custom type');
+
+# 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);
+ ));
+
+$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/);
+
+# Setting copy_binary to false should allow syncing
+$node_subscriber->safe_psql(
+ 'postgres', qq(
+ ALTER SUBSCRIPTION tsub SET (copy_binary = false);));
+
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tsub');
+
+$sync_check = qq(
+ SET timezone = '+2';
+ SELECT a, b, c, d FROM test_numerical ORDER BY a;
+ SELECT a, b, c FROM test_arrays ORDER BY a;
+ SELECT a, b FROM test_one_comp ORDER BY a;
+ SELECT a, b, c FROM test_range_array ORDER BY a;
+ SELECT a FROM test_myvarchar;
+ SELECT a FROM test_mismatching_types ORDER BY a;
+);
+
+# Check the synced data on subscribers
+$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}
+1|(1,a,1)
+2|(2,b,2)
+1|["2014-08-04 00:00:00+02",infinity)|{"[1,3)","[10,21)"}
+2|["2014-08-02 00:00:00+02","2014-08-04 00:00:00+02")|{"[2,4)","[20,31)"}
+a
+1
+2', 'check synced data on subscriber with copy_binary = false');
+
+$node_subscriber->stop('fast');
+$node_publisher->stop('fast');
+
+done_testing();
--
2.25.1
^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2023-02-28 10:59 vignesh C <[email protected]>
parent: Melih Mutlu <[email protected]>
1 sibling, 1 reply; 14+ messages in thread
From: vignesh C @ 2023-02-28 10:59 UTC (permalink / raw)
To: Melih Mutlu <[email protected]>; +Cc: Bharath Rupireddy <[email protected]>; Takamichi Osumi (Fujitsu) <[email protected]>; Jelte Fennema <[email protected]>; Amit Kapila <[email protected]>; pgsql-hackers; Hayato Kuroda (Fujitsu) <[email protected]>; [email protected] <[email protected]>; Euler Taveira <[email protected]>
On Tue, 28 Feb 2023 at 01:22, Melih Mutlu <[email protected]> wrote:
>
> Hi,
>
> Thanks for all of your reviews!
>
> So, I made some changes in the v10 according to your comments.
>
> 1- copy_format is changed to a boolean parameter copy_binary.
> It sounds easier to use a boolean to enable/disable binary copy. Default value is false, so nothing changes in the current behaviour if copy_binary is not specified.
> It's still persisted into the catalog. This is needed since its value will be needed by tablesync workers later. And tablesync workers fetch subscription configurations from the catalog.
> In the copy_data case, it is not directly stored anywhere but it affects the state of the table which is stored in the catalog. So, I guess this is the convenient way if we decide to go with a new parameter.
>
> 2- copy_binary is not tied to another parameter
> The patch does not disallow any combination of copy_binary with binary and copy_data options. I tried to explain what binary copy can and cannot do in the docs. Rest is up to the user now.
>
> 3- Removed version check for copy_binary
> HEAD already does not have any check for binary option. Making binary copy work only if publisher and subscriber are the same major version can be too restricted.
>
> 4- Added separate test file
> Although I believe 002_types.pl and 014_binary.pl can be improved more for binary enabled subscription cases, this patch might not be the best place to make those changes.
> 032_binary_copy.pl now has the tests for binary copy option. There are also some regression tests in subscription.sql.
>
> Finally, some other small fixes are done according to the reviews.
>
> Also, thanks Bharath for performance testing [1]. It can be seen that the patch has some benefits.
>
> I appreciate further thoughts/reviews.
Thanks for the patch, Few comments:
1) Are primary key required for the tables, if not required we could
remove the primary key which will speed up the test by not creating
the indexes and inserting in the indexes. Even if required just create
for one of the tables:
+# Create tables on both sides of the replication
+my $ddl = qq(
+ CREATE TABLE public.test_numerical (
+ a INTEGER PRIMARY KEY,
+ b NUMERIC,
+ c FLOAT,
+ d BIGINT
+ );
+ CREATE TABLE public.test_arrays (
+ a INTEGER[] PRIMARY KEY,
+ b NUMERIC[],
+ c TEXT[]
+ );
+ CREATE TABLE public.test_range_array (
+ a INTEGER PRIMARY KEY,
+ b TSTZRANGE,
+ c int8range[]
+ );
+ CREATE TYPE public.test_comp_basic_t AS (a FLOAT, b TEXT, c INTEGER);
+ CREATE TABLE public.test_one_comp (
+ a INTEGER PRIMARY KEY,
+ b public.test_comp_basic_t
+ ););
+
2) We can have the Insert/Select of tables in the same order so that
it is easy to verify. test_range_array/test_one_comp
insertion/selection order was different.
+# Insert some content and 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"}');
+ INSERT INTO test_range_array (a, b, c) VALUES
+ (1, tstzrange('Mon Aug 04 00:00:00 2014
CEST'::timestamptz, 'infinity'), '{"[1,2]", "[10,20]"}'),
+ (2, tstzrange('Sat Aug 02 00:00:00 2014
CEST'::timestamptz, 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz),
'{"[2,3]", "[20,30]"}');
+ INSERT INTO test_one_comp (a, b) VALUES
+ (1, ROW(1.0, 'a', 1)),
+ (2, ROW(2.0, 'b', 2));
+ ));
+
+# Create the subscription with copy_binary = true
+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, copy_binary
= true)");
+
+# Ensure nodes are in sync with each other
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tsub');
+
+my $sync_check = qq(
+ SET timezone = '+2';
+ SELECT a, b, c, d FROM test_numerical ORDER BY a;
+ SELECT a, b, c FROM test_arrays ORDER BY a;
+ SELECT a, b FROM test_one_comp ORDER BY a;
+ SELECT a, b, c FROM test_range_array ORDER BY a;
+);
3) Should we check the results for test_myvarchar table only, since
there is no change in other tables, we need not check other tables
again:
+# Now tablesync should succeed
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tsub');
+
+$sync_check = qq(
+ SET timezone = '+2';
+ SELECT a, b, c, d FROM test_numerical ORDER BY a;
+ SELECT a, b, c FROM test_arrays ORDER BY a;
+ SELECT a, b FROM test_one_comp ORDER BY a;
+ SELECT a, b, c FROM test_range_array ORDER BY a;
+ SELECT a FROM test_myvarchar;
+);
+
+# Check the synced data on subscriber
+$result = $node_subscriber->safe_psql('postgres', $sync_check);
4) Similarly check only for test_mismatching_types in this case:
+# Cannot sync due to type mismatch
+$node_subscriber->wait_for_log(qr/ERROR: ( [A-Z0-9]+:)? incorrect
binary data format/);
+
+# Setting copy_binary to false should allow syncing
+$node_subscriber->safe_psql(
+ 'postgres', qq(
+ ALTER SUBSCRIPTION tsub SET (copy_binary = false);));
+
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tsub');
+
+$sync_check = qq(
+ SET timezone = '+2';
+ SELECT a, b, c, d FROM test_numerical ORDER BY a;
+ SELECT a, b, c FROM test_arrays ORDER BY a;
+ SELECT a, b FROM test_one_comp ORDER BY a;
+ SELECT a, b, c FROM test_range_array ORDER BY a;
+ SELECT a FROM test_myvarchar;
+ SELECT a FROM test_mismatching_types ORDER BY a;
+);
+
+# Check the synced data on subscribers
+$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}
+1|(1,a,1)
+2|(2,b,2)
+1|["2014-08-04 00:00:00+02",infinity)|{"[1,3)","[10,21)"}
+2|["2014-08-02 00:00:00+02","2014-08-04 00:00:00+02")|{"[2,4)","[20,31)"}
+a
+1
+2', 'check synced data on subscriber with copy_binary = false');
5) Should we change "Basic logical replication test" to "Test logical
replication of copy table in binary"
diff --git a/src/test/subscription/t/032_binary_copy.pl
b/src/test/subscription/t/032_binary_copy.pl
new file mode 100644
index 0000000000..bcad66e5ea
--- /dev/null
+++ b/src/test/subscription/t/032_binary_copy.pl
@@ -0,0 +1,223 @@
+
+# Copyright (c) 2023, PostgreSQL Global Development Group
+
+# Basic logical replication test
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
6) We can change "Initialize publisher node" to "Create publisher
node" to maintain consistency.
+# Initialize publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->start;
+
+# Create subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init(allows_streaming => 'logical');
+$node_subscriber->start;
7) Should "Insert some content and before creating a subscription." be
changed to:
"Insert some content before creating a subscription."
+# Publish all tables
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tpub FOR ALL TABLES");
+
+# Insert some content and 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),
Regards,
Vignesh
^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2023-02-28 12:27 Bharath Rupireddy <[email protected]>
parent: Melih Mutlu <[email protected]>
1 sibling, 1 reply; 14+ messages in thread
From: Bharath Rupireddy @ 2023-02-28 12:27 UTC (permalink / raw)
To: Melih Mutlu <[email protected]>; +Cc: Takamichi Osumi (Fujitsu) <[email protected]>; Jelte Fennema <[email protected]>; Amit Kapila <[email protected]>; pgsql-hackers; Hayato Kuroda (Fujitsu) <[email protected]>; [email protected] <[email protected]>; Euler Taveira <[email protected]>
On Tue, Feb 28, 2023 at 1:22 AM Melih Mutlu <[email protected]> wrote:
>
> Hi,
>
> Thanks for all of your reviews!
>
> So, I made some changes in the v10 according to your comments.
Thanks. Some quick comments on v10:
1.
+ <para>
+ If true, initial data synchronization will be performed in binary format
+ </para></entry>
It's not just the initial table sync right? The table sync can happen
at any other point of time when ALTER SUBSCRIPTION ... REFRESH
PUBLICATION WITH (copy = true) is run.
How about - "If true, the subscriber requests publication for
pre-existing data in binary format"?
2.
+ Specifies whether pre-existing data on the publisher will be copied
+ to the subscriber in binary format. The default is
<literal>false</literal>.
+ Binary format is very data type specific, it will not allow copying
+ between different column types as opposed to text format. Note that
+ if this option is enabled, all data types which will be copied during
+ the initial synchronization should have binary send and
receive functions.
+ If this option is disabled, data format for the initial
synchronization
+ will be text.
Perhaps, this should cover the recommended cases for enabling this new
option - something like below (may not need to have exact wording, but
the recommended cases?):
"It is recommended to enable this option only when 1) the column data
types have appropriate binary send/receive functions, 2) not
replicating between different major versions or different platforms,
3) both publisher and subscriber tables have the exact same column
types (not when replicating from smallint to int or numeric to int8
and so on), 4) both publisher and subscriber supports COPY with binary
option, otherwise the table copy can fail."
3. I think the newly added tests must verify if the binary COPY is
picked up when enabled. Perhaps, looking at the publisher's server log
for 'COPY ... WITH BINARY format'? Maybe it's an overkill, otherwise,
we have no way of testing that the option took effect.
--
Bharath Rupireddy
PostgreSQL Contributors Team
RDS Open Source Databases
Amazon Web Services: https://aws.amazon.com
^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2023-02-28 13:25 Jelte Fennema <[email protected]>
parent: Bharath Rupireddy <[email protected]>
0 siblings, 0 replies; 14+ messages in thread
From: Jelte Fennema @ 2023-02-28 13:25 UTC (permalink / raw)
To: Bharath Rupireddy <[email protected]>; +Cc: Melih Mutlu <[email protected]>; Takamichi Osumi (Fujitsu) <[email protected]>; Amit Kapila <[email protected]>; pgsql-hackers; Hayato Kuroda (Fujitsu) <[email protected]>; [email protected] <[email protected]>; Euler Taveira <[email protected]>
> 3. I think the newly added tests must verify if the binary COPY is
> picked up when enabled. Perhaps, looking at the publisher's server log
> for 'COPY ... WITH BINARY format'? Maybe it's an overkill, otherwise,
> we have no way of testing that the option took effect.
Another way to test that BINARY is enabled could be to trigger one
of the failure cases.
^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2023-02-28 14:20 Melih Mutlu <[email protected]>
parent: vignesh C <[email protected]>
0 siblings, 0 replies; 14+ messages in thread
From: Melih Mutlu @ 2023-02-28 14:20 UTC (permalink / raw)
To: vignesh C <[email protected]>; Bharath Rupireddy <[email protected]>; Jelte Fennema <[email protected]>; +Cc: Takamichi Osumi (Fujitsu) <[email protected]>; Amit Kapila <[email protected]>; pgsql-hackers; Hayato Kuroda (Fujitsu) <[email protected]>; [email protected] <[email protected]>; Euler Taveira <[email protected]>
Hi,
Attached v11.
vignesh C <[email protected]>, 28 Şub 2023 Sal, 13:59 tarihinde şunu
yazdı:
> Thanks for the patch, Few comments:
> 1) Are primary key required for the tables, if not required we could
> remove the primary key which will speed up the test by not creating
> the indexes and inserting in the indexes. Even if required just create
> for one of the tables:
>
I think that having a primary key in tables for logical replication tests
is good for checking if log. rep. duplicates any row. Other tests also have
primary keys in almost all tables.
Bharath Rupireddy <[email protected]>, 28 Şub 2023
Sal, 15:27 tarihinde şunu yazdı:
> 1.
> + <para>
> + If true, initial data synchronization will be performed in binary
> format
> + </para></entry>
> It's not just the initial table sync right? The table sync can happen
> at any other point of time when ALTER SUBSCRIPTION ... REFRESH
> PUBLICATION WITH (copy = true) is run.
> How about - "If true, the subscriber requests publication for
> pre-existing data in binary format"?
>
I changed it as you suggested.
I sometimes feel like the phrase "initial sync" is used for initial sync of
a table, not a subscription. Or table syncs triggered by ALTER SUBSCRIPTION
are ignored in some places where "initial sync" is used.
2.
> Perhaps, this should cover the recommended cases for enabling this new
> option - something like below (may not need to have exact wording, but
> the recommended cases?):
> "It is recommended to enable this option only when 1) the column data
> types have appropriate binary send/receive functions, 2) not
> replicating between different major versions or different platforms,
> 3) both publisher and subscriber tables have the exact same column
> types (not when replicating from smallint to int or numeric to int8
> and so on), 4) both publisher and subscriber supports COPY with binary
> option, otherwise the table copy can fail."
>
I added a line stating that binary format is less portable across machine
architectures and versions as stated in COPY [1].
I don't think we should add line saying "recommended", but state the
restrictions clearly instead. It's also similar in COPY docs as well.
I think the explanation now covers all your points, right?
Jelte Fennema <[email protected]>, 28 Şub 2023 Sal, 16:25 tarihinde şunu
yazdı:
> > 3. I think the newly added tests must verify if the binary COPY is
> > picked up when enabled. Perhaps, looking at the publisher's server log
> > for 'COPY ... WITH BINARY format'? Maybe it's an overkill, otherwise,
> > we have no way of testing that the option took effect.
>
> Another way to test that BINARY is enabled could be to trigger one
> of the failure cases.
Yes, there is already a failure case for binary copy which resolves with
swithcing binary_copy to false.
But I also added checks for publisher logs now too.
[1] https://www.postgresql.org/docs/devel/sql-copy.html
Thanks,
--
Melih Mutlu
Microsoft
Attachments:
[application/octet-stream] v11-0001-Allow-logical-replication-to-copy-table-in-binary.patch (64.8K, ../../CAGPVpCSegnqzEJiCEPhXY3WFKfC_3=3h+hSCfc4=6qRAfmK+rQ@mail.gmail.com/3-v11-0001-Allow-logical-replication-to-copy-table-in-binary.patch)
download | inline diff:
From c64920bbe3f0fe243a3ffb001eae4a09cb5834e1 Mon Sep 17 00:00:00 2001
From: Melih Mutlu <[email protected]>
Date: Mon, 8 Aug 2022 14:14:44 +0300
Subject: [PATCH v11] Allow logical replication to copy table in binary
If binary option is enabled in a subscription and copy_format is set to binary, then copy tables in
binary format during table synchronization.
The patch introduces a new parameter, copy_binary, as subscription option to
allow users to choose the format of initial table synchronization
Without this patch, table are copied in text format even if the
subscription is created with binary option. When binary format is
beneficial to use, allowing the subscription to copy tables in binary in
table sync phase may reduce the time spent on copy depending on column types.
Discussion: https://postgr.es/m/CAGPVpCQvAziCLknEnygY0v1-KBtg%2BOm-9JHJYZOnNPKFJPompw%40mail.gmail.com
---
doc/src/sgml/catalogs.sgml | 10 +
doc/src/sgml/logical-replication.sgml | 9 +-
doc/src/sgml/ref/alter_subscription.sgml | 22 ++-
doc/src/sgml/ref/create_subscription.sgml | 18 ++
src/backend/catalog/pg_subscription.c | 1 +
src/backend/catalog/system_views.sql | 3 +-
src/backend/commands/subscriptioncmds.c | 57 +++++-
src/backend/replication/logical/tablesync.c | 10 +-
src/bin/pg_dump/pg_dump.c | 17 +-
src/bin/pg_dump/pg_dump.h | 1 +
src/bin/psql/describe.c | 11 +-
src/bin/psql/tab-complete.c | 14 +-
src/include/catalog/pg_subscription.h | 2 +
src/test/regress/expected/subscription.out | 178 ++++++++++--------
src/test/regress/sql/subscription.sql | 18 ++
src/test/subscription/meson.build | 1 +
src/test/subscription/t/032_binary_copy.pl | 193 ++++++++++++++++++++
17 files changed, 468 insertions(+), 97 deletions(-)
create mode 100644 src/test/subscription/t/032_binary_copy.pl
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index c1e4048054..e1b8029dd6 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -8002,6 +8002,16 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
origin.
</para></entry>
</row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>subcopybinary</structfield> <type>bool</type>
+ </para>
+ <para>
+ If true, the subscriber requests publication for pre-existing data in
+ binary format
+ </para></entry>
+ </row>
</tbody>
</tgroup>
</table>
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 1bd5660c87..83caf40a05 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -241,10 +241,11 @@
types of the columns do not need to match, as long as the text
representation of the data can be converted to the target type. For
example, you can replicate from a column of type <type>integer</type> to a
- 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.
+ column of type <type>bigint</type>. However, replication in binary format is
+ type specific and does not allow to replicate data between different types
+ according to its restrictions. 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.
</para>
<sect2 id="logical-replication-subscription-slot">
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 964fcbb8ff..737b55b2bf 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -179,6 +179,24 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
</para>
</listitem>
</varlistentry>
+
+ <varlistentry>
+ <term><literal>copy_binary</literal> (<type>boolean</type>)</term>
+ <listitem>
+ <para>
+ Specifies whether pre-existing data on the publisher will be copied
+ to the subscriber in binary format. The default is <literal>false</literal>.
+ Binary format can be faster than the text format for copying data, but
+ it is less portable across machine architectures and PostgreSQL versions.
+ Binary format is also very data type specific, it will not allow copying
+ between different column types as opposed to text format. Note that
+ if this option is enabled, all data types which will be copied during
+ the initial synchronization should have binary send and receive functions.
+ If this option is disabled, data format for the initial synchronization
+ will be text.
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist></para>
</listitem>
</varlistentry>
@@ -213,8 +231,8 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
are <literal>slot_name</literal>,
<literal>synchronous_commit</literal>,
<literal>binary</literal>, <literal>streaming</literal>,
- <literal>disable_on_error</literal>, and
- <literal>origin</literal>.
+ <literal>disable_on_error</literal>, <literal>origin</literal>
+ and <literal>copy_binary</literal>.
</para>
</listitem>
</varlistentry>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 51c45f17c7..87a64daf9a 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -349,6 +349,24 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
</para>
</listitem>
</varlistentry>
+
+ <varlistentry>
+ <term><literal>copy_binary</literal> (<type>boolean</type>)</term>
+ <listitem>
+ <para>
+ Specifies whether pre-existing data on the publisher will be copied
+ to the subscriber in binary format. The default is <literal>false</literal>.
+ Binary format can be faster than the text format for copying data, but
+ it is less portable across machine architectures and PostgreSQL versions.
+ Binary format is also very data type specific, it will not allow copying
+ between different column types as opposed to text format. Note that
+ if this option is enabled, all data types which will be copied during
+ the initial synchronization should have binary send and receive functions.
+ If this option is disabled, data format for the initial synchronization
+ will be text.
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist></para>
</listitem>
diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index a56ae311c3..ffd3324bc1 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -71,6 +71,7 @@ GetSubscription(Oid subid, bool missing_ok)
sub->stream = subform->substream;
sub->twophasestate = subform->subtwophasestate;
sub->disableonerr = subform->subdisableonerr;
+ sub->copybinary = subform->subcopybinary;
/* Get conninfo */
datum = SysCacheGetAttr(SUBSCRIPTIONOID,
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 34ca0e739f..2521606bbc 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1316,7 +1316,8 @@ REVOKE ALL ON pg_replication_origin_status FROM public;
REVOKE ALL ON pg_subscription FROM public;
GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled,
subbinary, substream, subtwophasestate, subdisableonerr,
- subslotname, subsynccommit, subpublications, suborigin)
+ subslotname, subsynccommit, subpublications, suborigin,
+ subcopybinary)
ON pg_subscription TO public;
CREATE VIEW pg_stat_subscription_stats AS
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 464db6d247..7c3d67ad73 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -66,6 +66,7 @@
#define SUBOPT_DISABLE_ON_ERR 0x00000400
#define SUBOPT_LSN 0x00000800
#define SUBOPT_ORIGIN 0x00001000
+#define SUBOPT_COPY_BINARY 0x00002000
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
@@ -90,6 +91,7 @@ typedef struct SubOpts
bool disableonerr;
char *origin;
XLogRecPtr lsn;
+ bool copy_binary;
} SubOpts;
static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
@@ -146,6 +148,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->disableonerr = false;
if (IsSet(supported_opts, SUBOPT_ORIGIN))
opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY);
+ if (IsSet(supported_opts, SUBOPT_COPY_BINARY))
+ opts->copy_binary = false;
/* Parse options */
foreach(lc, stmt_options)
@@ -324,6 +328,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->specified_opts |= SUBOPT_LSN;
opts->lsn = lsn;
}
+ else if (IsSet(supported_opts, SUBOPT_COPY_BINARY) &&
+ strcmp(defel->defname, "copy_binary") == 0)
+ {
+ if (IsSet(opts->specified_opts, SUBOPT_COPY_BINARY))
+ errorConflictingDefElem(defel, pstate);
+
+ opts->specified_opts |= SUBOPT_COPY_BINARY;
+ opts->copy_binary = defGetBoolean(defel);
+ }
else
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
@@ -560,7 +573,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
SUBOPT_SLOT_NAME | SUBOPT_COPY_DATA |
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
- SUBOPT_DISABLE_ON_ERR | SUBOPT_ORIGIN);
+ SUBOPT_DISABLE_ON_ERR | SUBOPT_ORIGIN |
+ SUBOPT_COPY_BINARY);
parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
/*
@@ -636,6 +650,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
LOGICALREP_TWOPHASE_STATE_PENDING :
LOGICALREP_TWOPHASE_STATE_DISABLED);
values[Anum_pg_subscription_subdisableonerr - 1] = BoolGetDatum(opts.disableonerr);
+ values[Anum_pg_subscription_subcopybinary - 1] = BoolGetDatum(opts.copy_binary);
values[Anum_pg_subscription_subconninfo - 1] =
CStringGetTextDatum(conninfo);
if (opts.slot_name)
@@ -1054,7 +1069,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
supported_opts = (SUBOPT_SLOT_NAME |
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
SUBOPT_STREAMING | SUBOPT_DISABLE_ON_ERR |
- SUBOPT_ORIGIN);
+ SUBOPT_ORIGIN | SUBOPT_COPY_BINARY);
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -1118,6 +1133,13 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
replaces[Anum_pg_subscription_suborigin - 1] = true;
}
+ if (IsSet(opts.specified_opts, SUBOPT_COPY_BINARY))
+ {
+ values[Anum_pg_subscription_subcopybinary - 1] =
+ BoolGetDatum(opts.copy_binary);
+ replaces[Anum_pg_subscription_subcopybinary - 1] = true;
+ }
+
update_tuple = true;
break;
}
@@ -1158,7 +1180,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
case ALTER_SUBSCRIPTION_SET_PUBLICATION:
{
- supported_opts = SUBOPT_COPY_DATA | SUBOPT_REFRESH;
+ supported_opts = (SUBOPT_COPY_DATA | SUBOPT_REFRESH |
+ SUBOPT_COPY_BINARY);
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -1166,6 +1189,13 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
publicationListToArray(stmt->publication);
replaces[Anum_pg_subscription_subpublications - 1] = true;
+ if (IsSet(opts.specified_opts, SUBOPT_COPY_BINARY))
+ {
+ values[Anum_pg_subscription_subcopybinary - 1] =
+ BoolGetDatum(opts.copy_binary);
+ replaces[Anum_pg_subscription_subcopybinary - 1] = true;
+ }
+
update_tuple = true;
/* Refresh if user asked us to. */
@@ -1205,7 +1235,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
List *publist;
bool isadd = stmt->kind == ALTER_SUBSCRIPTION_ADD_PUBLICATION;
- supported_opts = SUBOPT_REFRESH | SUBOPT_COPY_DATA;
+ supported_opts = (SUBOPT_REFRESH | SUBOPT_COPY_DATA |
+ SUBOPT_COPY_BINARY);
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -1214,6 +1245,13 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
publicationListToArray(publist);
replaces[Anum_pg_subscription_subpublications - 1] = true;
+ if (IsSet(opts.specified_opts, SUBOPT_COPY_BINARY))
+ {
+ values[Anum_pg_subscription_subcopybinary - 1] =
+ BoolGetDatum(opts.copy_binary);
+ replaces[Anum_pg_subscription_subcopybinary - 1] = true;
+ }
+
update_tuple = true;
/* Refresh if user asked us to. */
@@ -1265,8 +1303,17 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("ALTER SUBSCRIPTION ... REFRESH is not allowed for disabled subscriptions")));
+ supported_opts = SUBOPT_COPY_DATA | SUBOPT_COPY_BINARY;
parse_subscription_options(pstate, stmt->options,
- SUBOPT_COPY_DATA, &opts);
+ supported_opts, &opts);
+
+ if (IsSet(opts.specified_opts, SUBOPT_COPY_BINARY))
+ {
+ values[Anum_pg_subscription_subcopybinary - 1] =
+ BoolGetDatum(opts.copy_binary);
+ replaces[Anum_pg_subscription_subcopybinary - 1] = true;
+ }
+ update_tuple = true;
/*
* The subscription option "two_phase" requires that
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 07eea504ba..90f0507fe6 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -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"
@@ -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)),
@@ -1168,6 +1170,12 @@ copy_table(Relation rel)
appendStringInfoString(&cmd, ") TO STDOUT");
}
+
+ appendStringInfo(&cmd, " WITH (FORMAT %s)", MySubscription->copybinary ? "binary" : "text");
+ options = lappend(options, makeDefElem("format",
+ (Node *) makeString(MySubscription->copybinary ? "binary" : "text"),
+ -1));
+
res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 0, NULL);
pfree(cmd.data);
if (res->status != WALRCV_OK_COPY_OUT)
@@ -1184,7 +1192,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);
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 24ba936332..1a863464c7 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4493,6 +4493,7 @@ getSubscriptions(Archive *fout)
int i_subsynccommit;
int i_subpublications;
int i_subbinary;
+ int i_subcopybinary;
int i,
ntups;
@@ -4545,9 +4546,15 @@ getSubscriptions(Archive *fout)
LOGICALREP_TWOPHASE_STATE_DISABLED);
if (fout->remoteVersion >= 160000)
- appendPQExpBufferStr(query, " s.suborigin\n");
+ {
+ appendPQExpBufferStr(query, " s.suborigin,\n"
+ " s.subcopybinary\n");
+ }
else
- appendPQExpBuffer(query, " '%s' AS suborigin\n", LOGICALREP_ORIGIN_ANY);
+ {
+ appendPQExpBuffer(query, " '%s' AS suborigin,\n", LOGICALREP_ORIGIN_ANY);
+ appendPQExpBuffer(query, " false AS subcopybinary\n");
+ }
appendPQExpBufferStr(query,
"FROM pg_subscription s\n"
@@ -4575,6 +4582,7 @@ getSubscriptions(Archive *fout)
i_subtwophasestate = PQfnumber(res, "subtwophasestate");
i_subdisableonerr = PQfnumber(res, "subdisableonerr");
i_suborigin = PQfnumber(res, "suborigin");
+ i_subcopybinary = PQfnumber(res, "subcopybinary");
subinfo = pg_malloc(ntups * sizeof(SubscriptionInfo));
@@ -4605,6 +4613,8 @@ getSubscriptions(Archive *fout)
subinfo[i].subdisableonerr =
pg_strdup(PQgetvalue(res, i, i_subdisableonerr));
subinfo[i].suborigin = pg_strdup(PQgetvalue(res, i, i_suborigin));
+ subinfo[i].subcopybinary =
+ pg_strdup(PQgetvalue(res, i, i_subcopybinary));
/* Decide whether we want to dump it */
selectDumpableObject(&(subinfo[i].dobj), fout);
@@ -4686,6 +4696,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
if (strcmp(subinfo->subsynccommit, "off") != 0)
appendPQExpBuffer(query, ", synchronous_commit = %s", fmtId(subinfo->subsynccommit));
+ if (strcmp(subinfo->subcopybinary, "t") == 0)
+ appendPQExpBuffer(query, ", copy_binary = true");
+
appendPQExpBufferStr(query, ");\n");
if (subinfo->dobj.dump & DUMP_COMPONENT_DEFINITION)
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index cdca0b993d..dc1227e494 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -661,6 +661,7 @@ typedef struct _SubscriptionInfo
char *suborigin;
char *subsynccommit;
char *subpublications;
+ char *subcopybinary;
} SubscriptionInfo;
/*
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index c8a0bb7b3a..edace80acb 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -6472,7 +6472,7 @@ describeSubscriptions(const char *pattern, bool verbose)
PGresult *res;
printQueryOpt myopt = pset.popt;
static const bool translate_columns[] = {false, false, false, false,
- false, false, false, false, false, false, false, false};
+ false, false, false, false, false, false, false, false, false};
if (pset.sversion < 100000)
{
@@ -6527,10 +6527,15 @@ describeSubscriptions(const char *pattern, bool verbose)
gettext_noop("Two-phase commit"),
gettext_noop("Disable on error"));
+ /* Copy binary and origin are only supported in v16 and higher */
if (pset.sversion >= 160000)
+ {
appendPQExpBuffer(&buf,
- ", suborigin AS \"%s\"\n",
- gettext_noop("Origin"));
+ ", suborigin AS \"%s\"\n"
+ ", subcopybinary AS \"%s\"\n",
+ gettext_noop("Origin"),
+ gettext_noop("Copy binary"));
+ }
appendPQExpBuffer(&buf,
", subsynccommit AS \"%s\"\n"
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 5e1882eaea..9635eb2740 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1919,14 +1919,14 @@ psql_completion(const char *text, int start, int end)
/* ALTER SUBSCRIPTION <name> REFRESH PUBLICATION WITH ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) &&
TailMatches("REFRESH", "PUBLICATION", "WITH", "("))
- COMPLETE_WITH("copy_data");
+ COMPLETE_WITH("copy_binary", "copy_data");
/* ALTER SUBSCRIPTION <name> SET */
else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, "SET"))
COMPLETE_WITH("(", "PUBLICATION");
/* ALTER SUBSCRIPTION <name> SET ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "("))
- COMPLETE_WITH("binary", "disable_on_error", "origin", "slot_name",
- "streaming", "synchronous_commit");
+ COMPLETE_WITH("binary", "copy_binary", "disable_on_error", "origin",
+ "slot_name", "streaming", "synchronous_commit");
/* ALTER SUBSCRIPTION <name> SKIP ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SKIP", "("))
COMPLETE_WITH("lsn");
@@ -1942,7 +1942,7 @@ psql_completion(const char *text, int start, int end)
/* ALTER SUBSCRIPTION <name> ADD|DROP|SET PUBLICATION <name> WITH ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) &&
TailMatches("ADD|DROP|SET", "PUBLICATION", MatchAny, "WITH", "("))
- COMPLETE_WITH("copy_data", "refresh");
+ COMPLETE_WITH("copy_binary", "copy_data", "refresh");
/* ALTER SCHEMA <name> */
else if (Matches("ALTER", "SCHEMA", MatchAny))
@@ -3267,9 +3267,9 @@ psql_completion(const char *text, int start, int end)
COMPLETE_WITH("WITH (");
/* Complete "CREATE SUBSCRIPTION <name> ... WITH ( <opt>" */
else if (HeadMatches("CREATE", "SUBSCRIPTION") && TailMatches("WITH", "("))
- COMPLETE_WITH("binary", "connect", "copy_data", "create_slot",
- "disable_on_error", "enabled", "origin", "slot_name",
- "streaming", "synchronous_commit", "two_phase");
+ COMPLETE_WITH("binary", "connect", "copy_binary", "copy_data",
+ "create_slot", "disable_on_error", "enabled", "origin",
+ "slot_name", "streaming", "synchronous_commit", "two_phase");
/* CREATE TRIGGER --- is allowed inside CREATE SCHEMA, so use TailMatches */
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index b0f2a1705d..a93a3f3dab 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -88,6 +88,7 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
bool subdisableonerr; /* True if a worker error should cause the
* subscription to be disabled */
+ bool subcopybinary; /* True if initial sync will be performed in binary*/
#ifdef CATALOG_VARLEN /* variable-length fields start here */
/* Connection string to the publisher */
text subconninfo BKI_FORCE_NOT_NULL;
@@ -131,6 +132,7 @@ typedef struct Subscription
bool disableonerr; /* Indicates if the subscription should be
* automatically disabled if a worker error
* occurs */
+ bool copybinary; /* Indicates if initial sync will be in binary */
char *conninfo; /* Connection string to the publisher */
char *slotname; /* Name of the replication slot */
char *synccommit; /* Synchronous commit setting for worker */
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 3f99b14394..700bd108b0 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -114,18 +114,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | none | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | none | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub4 SET (origin = any);
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub3;
@@ -143,10 +143,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@@ -163,10 +163,10 @@ ERROR: unrecognized subscription parameter: "create_slot"
-- ok
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345');
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | off | dbname=regress_doesnotexist2 | 0/12345
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | f | off | dbname=regress_doesnotexist2 | 0/12345
(1 row)
-- ok - with lsn = NONE
@@ -175,10 +175,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE);
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0');
ERROR: invalid WAL location (LSN): 0/0
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | off | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | f | off | dbname=regress_doesnotexist2 | 0/0
(1 row)
BEGIN;
@@ -210,10 +210,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar);
ERROR: invalid value for parameter "synchronous_commit": "foobar"
HINT: Available values: local, remote_write, remote_apply, on, off.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+--------------------+------------------------------+----------
- regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | local | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+------------------------------+----------
+ regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | f | local | dbname=regress_doesnotexist2 | 0/0
(1 row)
-- rename back to keep the rest simple
@@ -247,19 +247,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (binary = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -271,27 +271,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication already exists
@@ -306,10 +306,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr
ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false);
ERROR: publication "testpub1" is already in subscription "regress_testsub"
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication used more than once
@@ -324,10 +324,10 @@ ERROR: publication "testpub3" is not in subscription "regress_testsub"
-- ok - delete publications
ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -363,10 +363,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
--fail - alter of two_phase option not supported.
@@ -375,10 +375,10 @@ ERROR: unrecognized subscription parameter: "two_phase"
-- but can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -388,10 +388,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -404,18 +404,52 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | t | any | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | t | any | f | off | dbname=regress_doesnotexist | 0/0
+(1 row)
+
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+-- fail - copy_binary must be boolean
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_binary = foo);
+ERROR: copy_binary requires a Boolean value
+-- now it works - create a subscription with copy_binary = true
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_binary = true);
+WARNING: subscription was created, but is not connected
+HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
+\dRs+
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | off | dbname=regress_doesnotexist | 0/0
+(1 row)
+
+-- alter copy_binary to false
+ALTER SUBSCRIPTION regress_testsub SET (copy_binary = false);
+\dRs+
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | f | off | dbname=regress_doesnotexist | 0/0
+(1 row)
+
+-- add publication with copy_binary = true
+ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1 WITH (refresh = false, copy_binary = true);
+\dRs+
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Copy binary | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+--------------------+--------+-----------+------------------+------------------+--------+-------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub,testpub1} | f | off | d | f | any | t | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index 7281f5fee2..f6b08cbb08 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -286,6 +286,24 @@ ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
DROP SUBSCRIPTION regress_testsub;
+-- fail - copy_binary must be boolean
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_binary = foo);
+
+-- now it works - create a subscription with copy_binary = true
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_binary = true);
+\dRs+
+
+-- alter copy_binary to false
+ALTER SUBSCRIPTION regress_testsub SET (copy_binary = false);
+\dRs+
+
+-- add publication with copy_binary = true
+ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1 WITH (refresh = false, copy_binary = true);
+\dRs+
+
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+
RESET SESSION AUTHORIZATION;
DROP ROLE regress_subscription_user;
DROP ROLE regress_subscription_user2;
diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build
index 3db0fdfd96..f9ab6eb7e1 100644
--- a/src/test/subscription/meson.build
+++ b/src/test/subscription/meson.build
@@ -38,6 +38,7 @@ tests += {
't/029_on_error.pl',
't/030_origin.pl',
't/031_column_list.pl',
+ 't/032_binary_copy.pl',
't/100_bugs.pl',
],
},
diff --git a/src/test/subscription/t/032_binary_copy.pl b/src/test/subscription/t/032_binary_copy.pl
new file mode 100644
index 0000000000..a21ff67e3e
--- /dev/null
+++ b/src/test/subscription/t/032_binary_copy.pl
@@ -0,0 +1,193 @@
+
+# Copyright (c) 2023, PostgreSQL Global Development Group
+
+# Test the binary copy for logical replication
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# Create and initialize a publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->start;
+
+# Create and initialize a subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init(allows_streaming => 'logical');
+$node_subscriber->start;
+
+# Create tables on both sides of the replication
+my $ddl = qq(
+ CREATE TABLE public.test_numerical (
+ a INTEGER PRIMARY KEY,
+ b NUMERIC,
+ c FLOAT,
+ d BIGINT
+ );
+ CREATE TABLE public.test_arrays (
+ a INTEGER[] PRIMARY KEY,
+ b NUMERIC[],
+ c TEXT[]
+ );
+ CREATE TABLE public.test_range_array (
+ a INTEGER PRIMARY KEY,
+ b TSTZRANGE,
+ c int8range[]
+ );
+ CREATE TYPE public.test_comp_basic_t AS (a FLOAT, b TEXT, c INTEGER);
+ CREATE TABLE public.test_one_comp (
+ a INTEGER PRIMARY KEY,
+ b public.test_comp_basic_t
+ ););
+
+$node_publisher->safe_psql('postgres', $ddl);
+$node_subscriber->safe_psql('postgres', $ddl);
+
+# Publish all tables
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tpub FOR ALL TABLES");
+
+# 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"}');
+ INSERT INTO test_range_array (a, b, c) VALUES
+ (1, tstzrange('Mon Aug 04 00:00:00 2014 CEST'::timestamptz, 'infinity'), '{"[1,2]", "[10,20]"}'),
+ (2, tstzrange('Sat Aug 02 00:00:00 2014 CEST'::timestamptz, 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[2,3]", "[20,30]"}');
+ INSERT INTO test_one_comp (a, b) VALUES
+ (1, ROW(1.0, 'a', 1)),
+ (2, ROW(2.0, 'b', 2));
+ ));
+
+# Create the subscription with copy_binary = true
+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, copy_binary = true)");
+
+# Ensure the COPY command is executed in binary format on the publisher
+$node_publisher->wait_for_log(qr/LOG: ( [a-z0-9]+:)? 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(
+ SET timezone = '+2';
+ SELECT a, b, c, d FROM test_numerical ORDER BY a;
+ SELECT a, b, c FROM test_arrays ORDER BY a;
+ SELECT a, b, c FROM test_range_array ORDER BY a;
+ SELECT a, b FROM test_one_comp ORDER BY a;
+);
+
+# Check the synced data on subscribers
+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}
+1|["2014-08-04 00:00:00+02",infinity)|{"[1,3)","[10,21)"}
+2|["2014-08-02 00:00:00+02","2014-08-04 00:00:00+02")|{"[2,4)","[20,31)"}
+1|(1,a,1)
+2|(2,b,2)', 'check synced data on subscriber');
+
+# 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');
+ ));
+
+# 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/);
+
+# 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 subscriber
+$result = $node_subscriber->safe_psql('postgres', 'SELECT a FROM test_myvarchar;');
+
+is( $result, 'a', 'check synced data on subscriber with custom type');
+
+# 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);
+ ));
+
+$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/);
+
+# Setting copy_binary to false should allow syncing
+$node_subscriber->safe_psql(
+ 'postgres', qq(
+ ALTER SUBSCRIPTION tsub SET (copy_binary = false);));
+
+# Ensure the COPY command is executed in text format on the publisher
+$node_publisher->wait_for_log(qr/LOG: ( [a-z0-9]+:)? COPY (.+)? TO STDOUT WITH \(FORMAT text\)/);
+
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tsub');
+
+# Check the synced data on subscribers
+$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 copy_binary = false');
+
+$node_subscriber->stop('fast');
+$node_publisher->stop('fast');
+
+done_testing();
--
2.25.1
^ permalink raw reply [nested|flat] 14+ messages in thread
* [PATCH v15 2/8] Row pattern recognition patch (parse/analysis).
@ 2024-03-28 10:30 Tatsuo Ishii <[email protected]>
0 siblings, 0 replies; 14+ messages in thread
From: Tatsuo Ishii @ 2024-03-28 10:30 UTC (permalink / raw)
---
src/backend/parser/parse_agg.c | 7 +
src/backend/parser/parse_clause.c | 296 +++++++++++++++++++++++++++++-
src/backend/parser/parse_expr.c | 4 +
src/backend/parser/parse_func.c | 3 +
4 files changed, 309 insertions(+), 1 deletion(-)
diff --git a/src/backend/parser/parse_agg.c b/src/backend/parser/parse_agg.c
index bee7d8346a..9bc22a836a 100644
--- a/src/backend/parser/parse_agg.c
+++ b/src/backend/parser/parse_agg.c
@@ -577,6 +577,10 @@ check_agglevels_and_constraints(ParseState *pstate, Node *expr)
errkind = true;
break;
+ case EXPR_KIND_RPR_DEFINE:
+ errkind = true;
+ break;
+
/*
* There is intentionally no default: case here, so that the
* compiler will warn if we add a new ParseExprKind without
@@ -967,6 +971,9 @@ transformWindowFuncCall(ParseState *pstate, WindowFunc *wfunc,
case EXPR_KIND_CYCLE_MARK:
errkind = true;
break;
+ case EXPR_KIND_RPR_DEFINE:
+ errkind = true;
+ break;
/*
* There is intentionally no default: case here, so that the
diff --git a/src/backend/parser/parse_clause.c b/src/backend/parser/parse_clause.c
index d2ac86777c..de2e5791e3 100644
--- a/src/backend/parser/parse_clause.c
+++ b/src/backend/parser/parse_clause.c
@@ -98,7 +98,14 @@ static WindowClause *findWindowClause(List *wclist, const char *name);
static Node *transformFrameOffset(ParseState *pstate, int frameOptions,
Oid rangeopfamily, Oid rangeopcintype, Oid *inRangeFunc,
Node *clause);
-
+static void transformRPR(ParseState *pstate, WindowClause *wc, WindowDef *windef,
+ List **targetlist);
+static List *transformDefineClause(ParseState *pstate, WindowClause *wc, WindowDef *windef,
+ List **targetlist);
+static void transformPatternClause(ParseState *pstate, WindowClause *wc,
+ WindowDef *windef);
+static List *transformMeasureClause(ParseState *pstate, WindowClause *wc,
+ WindowDef *windef);
/*
* transformFromClause -
@@ -2948,6 +2955,10 @@ transformWindowDefinitions(ParseState *pstate,
rangeopfamily, rangeopcintype,
&wc->endInRangeFunc,
windef->endOffset);
+
+ /* Process Row Pattern Recognition related clauses */
+ transformRPR(pstate, wc, windef, targetlist);
+
wc->runCondition = NIL;
wc->winref = winref;
@@ -3813,3 +3824,286 @@ transformFrameOffset(ParseState *pstate, int frameOptions,
return node;
}
+
+/*
+ * transformRPR
+ * Process Row Pattern Recognition related clauses
+ */
+static void
+transformRPR(ParseState *pstate, WindowClause *wc, WindowDef *windef,
+ List **targetlist)
+{
+ /*
+ * Window definition exists?
+ */
+ if (windef == NULL)
+ return;
+
+ /*
+ * Row Pattern Common Syntax clause exists?
+ */
+ if (windef->rpCommonSyntax == NULL)
+ return;
+
+ /* Check Frame option. Frame must start at current row */
+ if ((wc->frameOptions & FRAMEOPTION_START_CURRENT_ROW) == 0)
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("FRAME must start at current row when row patttern recognition is used")));
+
+ /* Transform AFTER MACH SKIP TO clause */
+ wc->rpSkipTo = windef->rpCommonSyntax->rpSkipTo;
+
+ /* Transform AFTER MACH SKIP TO variable */
+ wc->rpSkipVariable = windef->rpCommonSyntax->rpSkipVariable;
+
+ /* Transform SEEK or INITIAL clause */
+ wc->initial = windef->rpCommonSyntax->initial;
+
+ /* Transform DEFINE clause into list of TargetEntry's */
+ wc->defineClause = transformDefineClause(pstate, wc, windef, targetlist);
+
+ /* Check PATTERN clause and copy to patternClause */
+ transformPatternClause(pstate, wc, windef);
+
+ /* Transform MEASURE clause */
+ transformMeasureClause(pstate, wc, windef);
+}
+
+/*
+ * transformDefineClause Process DEFINE clause and transform ResTarget into
+ * list of TargetEntry.
+ *
+ * XXX we only support column reference in row pattern definition search
+ * condition, e.g. "price". <row pattern definition variable name>.<column
+ * reference> is not supported, e.g. "A.price".
+ */
+static List *
+transformDefineClause(ParseState *pstate, WindowClause *wc, WindowDef *windef,
+ List **targetlist)
+{
+ /* DEFINE variable name initials */
+ static char *defineVariableInitials = "abcdefghijklmnopqrstuvwxyz";
+
+ ListCell *lc,
+ *l;
+ ResTarget *restarget,
+ *r;
+ List *restargets;
+ List *defineClause;
+ char *name;
+ int initialLen;
+ int i;
+
+ /*
+ * If Row Definition Common Syntax exists, DEFINE clause must exist. (the
+ * raw parser should have already checked it.)
+ */
+ Assert(windef->rpCommonSyntax->rpDefs != NULL);
+
+ /*
+ * Check and add "A AS A IS TRUE" if pattern variable is missing in DEFINE
+ * per the SQL standard.
+ */
+ restargets = NIL;
+ foreach(lc, windef->rpCommonSyntax->rpPatterns)
+ {
+ A_Expr *a;
+ bool found = false;
+
+ if (!IsA(lfirst(lc), A_Expr))
+ ereport(ERROR,
+ errmsg("node type is not A_Expr"));
+
+ a = (A_Expr *) lfirst(lc);
+ name = strVal(a->lexpr);
+
+ foreach(l, windef->rpCommonSyntax->rpDefs)
+ {
+ restarget = (ResTarget *) lfirst(l);
+
+ if (!strcmp(restarget->name, name))
+ {
+ found = true;
+ break;
+ }
+ }
+
+ if (!found)
+ {
+ /*
+ * "name" is missing. So create "name AS name IS TRUE" ResTarget
+ * node and add it to the temporary list.
+ */
+ A_Const *n;
+
+ restarget = makeNode(ResTarget);
+ n = makeNode(A_Const);
+ n->val.boolval.type = T_Boolean;
+ n->val.boolval.boolval = true;
+ n->location = -1;
+ restarget->name = pstrdup(name);
+ restarget->indirection = NIL;
+ restarget->val = (Node *) n;
+ restarget->location = -1;
+ restargets = lappend((List *) restargets, restarget);
+ }
+ }
+
+ if (list_length(restargets) >= 1)
+ {
+ /* add missing DEFINEs */
+ windef->rpCommonSyntax->rpDefs =
+ list_concat(windef->rpCommonSyntax->rpDefs, restargets);
+ list_free(restargets);
+ }
+
+ /*
+ * Check for duplicate row pattern definition variables. The standard
+ * requires that no two row pattern definition variable names shall be
+ * equivalent.
+ */
+ restargets = NIL;
+ foreach(lc, windef->rpCommonSyntax->rpDefs)
+ {
+ restarget = (ResTarget *) lfirst(lc);
+ name = restarget->name;
+
+ /*
+ * Add DEFINE expression (Restarget->val) to the targetlist as a
+ * TargetEntry if it does not exist yet. Planner will add the column
+ * ref var node to the outer plan's target list later on. This makes
+ * DEFINE expression could access the outer tuple while evaluating
+ * PATTERN.
+ *
+ * XXX: adding whole expressions of DEFINE to the plan.targetlist is
+ * not so good, because it's not necessary to evalute the expression
+ * in the target list while running the plan. We should extract the
+ * var nodes only then add them to the plan.targetlist.
+ */
+ findTargetlistEntrySQL99(pstate, (Node *) restarget->val,
+ targetlist, EXPR_KIND_RPR_DEFINE);
+
+ /*
+ * Make sure that the row pattern definition search condition is a
+ * boolean expression.
+ */
+ transformWhereClause(pstate, restarget->val,
+ EXPR_KIND_RPR_DEFINE, "DEFINE");
+
+ foreach(l, restargets)
+ {
+ char *n;
+
+ r = (ResTarget *) lfirst(l);
+ n = r->name;
+
+ if (!strcmp(n, name))
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("row pattern definition variable name \"%s\" appears more than once in DEFINE clause",
+ name),
+ parser_errposition(pstate, exprLocation((Node *) r))));
+ }
+ restargets = lappend(restargets, restarget);
+ }
+ list_free(restargets);
+
+ /*
+ * Create list of row pattern DEFINE variable name's initial. We assign
+ * [a-z] to them (up to 26 variable names are allowed).
+ */
+ restargets = NIL;
+ i = 0;
+ initialLen = strlen(defineVariableInitials);
+
+ foreach(lc, windef->rpCommonSyntax->rpDefs)
+ {
+ char initial[2];
+
+ restarget = (ResTarget *) lfirst(lc);
+ name = restarget->name;
+
+ if (i >= initialLen)
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("number of row pattern definition variable names exceeds %d",
+ initialLen),
+ parser_errposition(pstate,
+ exprLocation((Node *) restarget))));
+ }
+ initial[0] = defineVariableInitials[i++];
+ initial[1] = '\0';
+ wc->defineInitial = lappend(wc->defineInitial,
+ makeString(pstrdup(initial)));
+ }
+
+ defineClause = transformTargetList(pstate, windef->rpCommonSyntax->rpDefs,
+ EXPR_KIND_RPR_DEFINE);
+
+ /* mark column origins */
+ markTargetListOrigins(pstate, defineClause);
+
+ /* mark all nodes in the DEFINE clause tree with collation information */
+ assign_expr_collations(pstate, (Node *) defineClause);
+
+ return defineClause;
+}
+
+/*
+ * transformPatternClause
+ * Process PATTERN clause and return PATTERN clause in the raw parse tree
+ */
+static void
+transformPatternClause(ParseState *pstate, WindowClause *wc,
+ WindowDef *windef)
+{
+ ListCell *lc;
+
+ /*
+ * Row Pattern Common Syntax clause exists?
+ */
+ if (windef->rpCommonSyntax == NULL)
+ return;
+
+ wc->patternVariable = NIL;
+ wc->patternRegexp = NIL;
+ foreach(lc, windef->rpCommonSyntax->rpPatterns)
+ {
+ A_Expr *a;
+ char *name;
+ char *regexp;
+
+ if (!IsA(lfirst(lc), A_Expr))
+ ereport(ERROR,
+ errmsg("node type is not A_Expr"));
+
+ a = (A_Expr *) lfirst(lc);
+ name = strVal(a->lexpr);
+
+ wc->patternVariable = lappend(wc->patternVariable, makeString(pstrdup(name)));
+ regexp = strVal(lfirst(list_head(a->name)));
+
+ wc->patternRegexp = lappend(wc->patternRegexp, makeString(pstrdup(regexp)));
+ }
+}
+
+/*
+ * transformMeasureClause
+ * Process MEASURE clause
+ * XXX MEASURE clause is not supported yet
+ */
+static List *
+transformMeasureClause(ParseState *pstate, WindowClause *wc,
+ WindowDef *windef)
+{
+ if (windef->rowPatternMeasures == NIL)
+ return NIL;
+
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("%s", "MEASURE clause is not supported yet"),
+ parser_errposition(pstate, exprLocation((Node *) windef->rowPatternMeasures))));
+ return NIL;
+}
diff --git a/src/backend/parser/parse_expr.c b/src/backend/parser/parse_expr.c
index 73c83cea4a..8b0cc608bc 100644
--- a/src/backend/parser/parse_expr.c
+++ b/src/backend/parser/parse_expr.c
@@ -578,6 +578,7 @@ transformColumnRef(ParseState *pstate, ColumnRef *cref)
case EXPR_KIND_COPY_WHERE:
case EXPR_KIND_GENERATED_COLUMN:
case EXPR_KIND_CYCLE_MARK:
+ case EXPR_KIND_RPR_DEFINE:
/* okay */
break;
@@ -1817,6 +1818,7 @@ transformSubLink(ParseState *pstate, SubLink *sublink)
case EXPR_KIND_VALUES:
case EXPR_KIND_VALUES_SINGLE:
case EXPR_KIND_CYCLE_MARK:
+ case EXPR_KIND_RPR_DEFINE:
/* okay */
break;
case EXPR_KIND_CHECK_CONSTRAINT:
@@ -3197,6 +3199,8 @@ ParseExprKindName(ParseExprKind exprKind)
return "GENERATED AS";
case EXPR_KIND_CYCLE_MARK:
return "CYCLE";
+ case EXPR_KIND_RPR_DEFINE:
+ return "DEFINE";
/*
* There is intentionally no default: case here, so that the
diff --git a/src/backend/parser/parse_func.c b/src/backend/parser/parse_func.c
index 0cbc950c95..ad982a7c17 100644
--- a/src/backend/parser/parse_func.c
+++ b/src/backend/parser/parse_func.c
@@ -2657,6 +2657,9 @@ check_srf_call_placement(ParseState *pstate, Node *last_srf, int location)
case EXPR_KIND_CYCLE_MARK:
errkind = true;
break;
+ case EXPR_KIND_RPR_DEFINE:
+ errkind = true;
+ break;
/*
* There is intentionally no default: case here, so that the
--
2.25.1
----Next_Part(Thu_Mar_28_19_59_25_2024_076)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
filename="v15-0003-Row-pattern-recognition-patch-rewriter.patch"
^ permalink raw reply [nested|flat] 14+ messages in thread
end of thread, other threads:[~2024-03-28 10:30 UTC | newest]
Thread overview: 14+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2023-02-20 07:12 RE: Allow logical replication to copy tables in binary format Hayato Kuroda (Fujitsu) <[email protected]>
2023-02-20 11:46 ` Melih Mutlu <[email protected]>
2023-02-21 13:48 ` Bharath Rupireddy <[email protected]>
2023-02-27 13:24 ` Bharath Rupireddy <[email protected]>
2023-02-27 19:52 ` Melih Mutlu <[email protected]>
2023-02-28 10:59 ` vignesh C <[email protected]>
2023-02-28 14:20 ` Melih Mutlu <[email protected]>
2023-02-28 12:27 ` Bharath Rupireddy <[email protected]>
2023-02-28 13:25 ` Jelte Fennema <[email protected]>
2023-02-22 07:01 ` Hayato Kuroda (Fujitsu) <[email protected]>
2023-02-24 03:02 ` Peter Smith <[email protected]>
2023-02-27 10:16 ` Bharath Rupireddy <[email protected]>
2023-02-26 22:58 ` Takamichi Osumi (Fujitsu) <[email protected]>
2024-03-28 10:30 [PATCH v15 2/8] Row pattern recognition patch (parse/analysis). Tatsuo Ishii <[email protected]>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox