public inbox for [email protected]
help / color / mirror / Atom feedFrom: Melih Mutlu <[email protected]>
To: Amit Kapila <[email protected]>
Cc: pgsql-hackers <[email protected]>
Subject: Re: [PATCH] Reuse Workers and Replication Slots during Logical Replication
Date: Thu, 15 Dec 2022 15:03:16 +0300
Message-ID: <CAGPVpCQdZ_oj-QFcTOhTrUTs-NCKrrZ=ZNCNPR1qe27rXV-iYw@mail.gmail.com> (raw)
In-Reply-To: <CAGPVpCRWEVhXa7ovrhuSQofx4to7o22oU9iKtrOgAOtz_=Y6vg@mail.gmail.com>
References: <CAGPVpCTq=rUDd4JUdaRc1XUWf4BrH2gdSNf3rtOMUGj9rPpfzQ@mail.gmail.com>
<CAA4eK1Knrj73YW181CBW15vCAqEgNi7Xk7dG8Oah4gg0A_GcOA@mail.gmail.com>
<CAFiTN-tff+RVLwkx0DkM77YZniohoq9gJ5AByU-0B=KUCmUyjQ@mail.gmail.com>
<CAA4eK1JWJtQdoHEnRO5XYpXSv6YwdZ5KjUa5yV1onxGAtsL+MQ@mail.gmail.com>
<CAGPVpCTV57_aji6y4k1PwEVO3h8h2OYkEKiYC4OB=J-ErXufQw@mail.gmail.com>
<CAA4eK1LrUX5F1Bj8vAW-tvVt3isn+xVj8JdRxQA2xeCYz0FB8w@mail.gmail.com>
<CAGPVpCTKqBZxkAj4rTqwzdH034fUfwfUF1fHTTsNrt4+gkJJZQ@mail.gmail.com>
<CAA4eK1Lo4A4Anyp07r+XF4zp7okfNtUiM7_=e_s7tUedUB3ejw@mail.gmail.com>
<CAGPVpCRdDBRav4AR6NgAg+7HBfokppWwJkmjwzjWeuy1i7HYqA@mail.gmail.com>
<CAA4eK1+PsvbjSH7yiCdCn=zcpAdM01MK=wdOuq879JcQqu6jwg@mail.gmail.com>
<CAGPVpCQazGgGu60YXfdoUqqjn9-B7vP7=DTNCZfp51Fg==pQ9w@mail.gmail.com>
<CAA4eK1KdXirEBrEhU38AHnMpa1NKMM40TBQ7qPdveJ4BS5KtDw@mail.gmail.com>
<CAGPVpCT0+_np5z9jj8OcFNtzoeSSoT+fObNQNFDjPeB12HMinw@mail.gmail.com>
<CAA4eK1Jv1hgMFkh896N6gPdDrOvKiAjnTNTsbVZL+OhUudZJ0Q@mail.gmail.com>
<CAGPVpCRWEVhXa7ovrhuSQofx4to7o22oU9iKtrOgAOtz_=Y6vg@mail.gmail.com>
Hi,
Attached new versions of the patch with some changes/fixes.
Here also some numbers to compare the performance of log. rep. with this
patch against the current master branch.
My method of benchmarking is the same with what I did earlier in this
thread. (on a different environment, so not compare the result from this
email with the ones from earlier emails)
> With those changes, I did some benchmarking to see if it improves anything.
> This results compares this patch with the latest version of master branch.
> "max_sync_workers_per_subscription" is set to 2 as default.
> Got some results simply averaging timings from 5 consecutive runs for each
> branch.
Since this patch is expected to improve log. rep. of empty/close-to-empty
tables, started with measuring performance with empty tables.
| 10 tables | 100 tables | 1000 tables
------------------------------------------------------------------------------
master | 283.430 ms | 22739.107 ms | 105226.177 ms
------------------------------------------------------------------------------
patch | 189.139 ms | 1554.802 ms | 23091.434 ms
After the changes discussed here [1], concurrent replication origin drops
> by apply worker and tablesync workers may hold each other on wait due to
> locks taken by replorigin_drop_by_name.
> I see that this harms the performance of logical replication quite a bit
> in terms of speed.
> [1]
> https://www.postgresql.org/message-id/flat/20220714115155.GA5439%40depesz.com
Firstly, as I mentioned, replication origin drops made things worse for the
master branch.
Locks start being a more serious issue when the number of tables increases.
The patch reuses the origin so does not need to drop them in each
iteration. That's why the difference between the master and the patch is
more significant now than it was when I first sent the patch.
To just show that the improvement is not only the result of reuse of
origins, but also reuse of rep. slots and workers, I just reverted those
commits which causes the origin drop issue.
| 10 tables | 100 tables | 1000 tables
-----------------------------------------------------------------------------
reverted | 270.012 ms | 2483.907 ms | 31660.758 ms
-----------------------------------------------------------------------------
patch | 189.139 ms | 1554.802 ms | 23091.434 ms
With this patch, logical replication is still faster, even if we wouldn't
have an issue with rep. origin drops.
Also here are some numbers with 10 tables loaded with some data :
| 10 MB | 100 MB
----------------------------------------------------------
master | 2868.524 ms | 14281.711 ms
----------------------------------------------------------
patch | 1750.226 ms | 14592.800 ms
The difference between the master and the patch is getting close when the
size of tables increase, as expected.
I would appreciate any feedback/thought on the approach/patch/numbers etc.
Thanks,
--
Melih Mutlu
Microsoft
Attachments:
[application/octet-stream] v2-0001-Add-replication-protocol-cmd-to-create-a-snapshot.patch (20.5K, ../CAGPVpCQdZ_oj-QFcTOhTrUTs-NCKrrZ=ZNCNPR1qe27rXV-iYw@mail.gmail.com/3-v2-0001-Add-replication-protocol-cmd-to-create-a-snapshot.patch)
download | inline diff:
From 9ffa22049582ac0464306d301b57aba8610376e7 Mon Sep 17 00:00:00 2001
From: Melih Mutlu <[email protected]>
Date: Thu, 13 Oct 2022 17:05:45 +0300
Subject: [PATCH 1/2] Add replication protocol cmd to create a snapshot
Introduced REPLICATION_SLOT_SNAPSHOT to be able to create and use a
snapshot without creating a new replication slot, but by using an
existing slot.
REPLICATION_SLOT_SNAPSHOT simply does what CREATE_REPLICATION_SLOT does
without creating a new replication slot.
REPLICATION_SLOT_SNAPSHOT command imports the snapshot into the current
transaction and returns consistent_point. The changes earlier than the
consistent_point will be applied by importing the snapshot. All changes
later than the consistent_point will be available to be consumed from
the replication slot.
This is useful for reusing replication slots in logical replication.
Otherwise, tablesync workers cannot start from a consistent point to copy a relation and then apply changes by consuming from replication slot.
---
doc/src/sgml/protocol.sgml | 32 ++++++
.../libpqwalreceiver/libpqwalreceiver.c | 69 ++++++++++++-
src/backend/replication/logical/logical.c | 39 +++++++-
.../replication/logical/logicalfuncs.c | 1 +
src/backend/replication/repl_gram.y | 18 +++-
src/backend/replication/repl_scanner.l | 2 +
src/backend/replication/slotfuncs.c | 1 +
src/backend/replication/walsender.c | 97 ++++++++++++++++++-
src/include/nodes/replnodes.h | 11 +++
src/include/replication/logical.h | 1 +
src/include/replication/walreceiver.h | 13 +++
src/tools/pgindent/typedefs.list | 1 +
12 files changed, 281 insertions(+), 4 deletions(-)
diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index 03312e07e2..f4f29079ac 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -2613,6 +2613,38 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
</listitem>
</varlistentry>
+ <varlistentry id="protocol-replication-replication-slot-snapshot">
+ <term><literal>REPLICATION_SLOT_SNAPSHOT</literal> <replaceable class="parameter">slot_name</replaceable> [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ]
+ <indexterm><primary>REPLICATION_SLOT_SNAPSHOT</primary></indexterm>
+ </term>
+ <listitem>
+ <para>
+ Creates a snapshot including all the changes from the replication slot until
+ the point at which the replication slot becomes consistent. Then the snapshot
+ is used in the currenct transaction. This command is currently only supported
+ for logical replication.
+ slots.
+ </para>
+
+ <para>
+ In response to this command, the server will return a one-row result set,
+ containing the following field:
+ <variablelist>
+ <varlistentry>
+ <term><literal>consistent_point</literal> (<type>text</type>)</term>
+ <listitem>
+ <para>
+ The WAL location at which the slot became consistent. This is the
+ earliest location from which streaming can start on this replication
+ slot.
+ </para>
+ </listitem>
+ </varlistentry>
+ </variablelist>
+ </para>
+ </listitem>
+ </varlistentry>
+
<varlistentry id="protocol-replication-base-backup" xreflabel="BASE_BACKUP">
<term><literal>BASE_BACKUP</literal> [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ]
<indexterm><primary>BASE_BACKUP</primary></indexterm>
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 7f697b0f29..e367c5f757 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -80,6 +80,8 @@ static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
const int nRetTypes,
const Oid *retTypes);
static void libpqrcv_disconnect(WalReceiverConn *conn);
+static void libpqrcv_slot_snapshot(WalReceiverConn *conn, char *slotname,
+ const WalRcvStreamOptions *options, XLogRecPtr *lsn);
static WalReceiverFunctionsType PQWalReceiverFunctions = {
.walrcv_connect = libpqrcv_connect,
@@ -96,7 +98,8 @@ static WalReceiverFunctionsType PQWalReceiverFunctions = {
.walrcv_create_slot = libpqrcv_create_slot,
.walrcv_get_backend_pid = libpqrcv_get_backend_pid,
.walrcv_exec = libpqrcv_exec,
- .walrcv_disconnect = libpqrcv_disconnect
+ .walrcv_disconnect = libpqrcv_disconnect,
+ .walrcv_slot_snapshot = libpqrcv_slot_snapshot
};
/* Prototypes for private functions */
@@ -968,6 +971,70 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
return snapshot;
}
+/*
+ * TODO
+ */
+static void
+libpqrcv_slot_snapshot(WalReceiverConn *conn,
+ char *slotname,
+ const WalRcvStreamOptions *options,
+ XLogRecPtr *lsn)
+{
+ StringInfoData cmd;
+ PGresult *res;
+ char *pubnames_str;
+ List *pubnames;
+ char *pubnames_literal;
+
+ initStringInfo(&cmd);
+
+ /* Build the command. */
+ appendStringInfo(&cmd, "REPLICATION_SLOT_SNAPSHOT \"%s\"", slotname);
+ appendStringInfoString(&cmd, " (");
+ appendStringInfo(&cmd, " proto_version '%u'",
+ options->proto.logical.proto_version);
+
+ /* Add publication names. */
+ pubnames = options->proto.logical.publication_names;
+ pubnames_str = stringlist_to_identifierstr(conn->streamConn, pubnames);
+ if (!pubnames_str)
+ ereport(ERROR,
+ (errcode(ERRCODE_OUT_OF_MEMORY), /* likely guess */
+ errmsg("could not start WAL streaming: %s",
+ pchomp(PQerrorMessage(conn->streamConn)))));
+ pubnames_literal = PQescapeLiteral(conn->streamConn, pubnames_str,
+ strlen(pubnames_str));
+ if (!pubnames_literal)
+ ereport(ERROR,
+ (errcode(ERRCODE_OUT_OF_MEMORY), /* likely guess */
+ errmsg("could not start WAL streaming: %s",
+ pchomp(PQerrorMessage(conn->streamConn)))));
+ appendStringInfo(&cmd, ", publication_names %s", pubnames_literal);
+ PQfreemem(pubnames_literal);
+ pfree(pubnames_str);
+
+ appendStringInfoString(&cmd, " )");
+
+ /* Execute the command. */
+ res = libpqrcv_PQexec(conn->streamConn, cmd.data);
+ pfree(cmd.data);
+
+ if (PQresultStatus(res) != PGRES_TUPLES_OK)
+ {
+ PQclear(res);
+ ereport(ERROR,
+ (errcode(ERRCODE_PROTOCOL_VIOLATION),
+ errmsg("Could not create a snapshot by replication slot \"%s\": %s",
+ slotname, pchomp(PQerrorMessage(conn->streamConn)))));
+ }
+
+ if (lsn)
+ *lsn = DatumGetLSN(DirectFunctionCall1Coll(pg_lsn_in, InvalidOid,
+ CStringGetDatum(PQgetvalue(res, 0, 0))));
+
+ PQclear(res);
+}
+
/*
* Return PID of remote backend process.
*/
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 625a7f4273..16c0a27ba5 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -461,6 +461,10 @@ CreateInitDecodingContext(const char *plugin,
* fast_forward
* bypass the generation of logical changes.
*
+ * need_full_snapshot
+ * if true, create a snapshot able to read all tables,
+ * otherwise do not create any snapshot.
+ *
* xl_routine
* XLogReaderRoutine used by underlying xlogreader
*
@@ -479,6 +483,7 @@ LogicalDecodingContext *
CreateDecodingContext(XLogRecPtr start_lsn,
List *output_plugin_options,
bool fast_forward,
+ bool need_full_snapshot,
XLogReaderRoutine *xl_routine,
LogicalOutputPluginWriterPrepareWrite prepare_write,
LogicalOutputPluginWriterWrite do_write,
@@ -487,6 +492,7 @@ CreateDecodingContext(XLogRecPtr start_lsn,
LogicalDecodingContext *ctx;
ReplicationSlot *slot;
MemoryContext old_context;
+ TransactionId xmin_horizon = InvalidTransactionId;
/* shorter lines... */
slot = MyReplicationSlot;
@@ -533,8 +539,39 @@ CreateDecodingContext(XLogRecPtr start_lsn,
start_lsn = slot->data.confirmed_flush;
}
+
+ /*
+ * We need to determine a safe xmin horizon to start decoding from if we
+ * want to create a snapshot too. Otherwise we would end up with a
+ * snapshot that cannot be imported since xmin value from the snapshot may
+ * be less than the oldest safe xmin. To avoid this call
+ * GetOldestSafeDecodingTransactionId() to return a safe xmin value, which
+ * can be used while exporting/importing the snapshot.
+ *
+ * So we have to acquire the ProcArrayLock to prevent computation of new
+ * xmin horizons by other backends, get the safe decoding xid, and inform
+ * the slot machinery about the new limit. Once that's done the
+ * ProcArrayLock can be released as the slot machinery now is protecting
+ * against vacuum.
+ */
+ if (need_full_snapshot)
+ {
+ LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+
+ SpinLockAcquire(&slot->mutex);
+ slot->effective_catalog_xmin = xmin_horizon;
+ slot->data.catalog_xmin = xmin_horizon;
+ slot->effective_xmin = xmin_horizon;
+ SpinLockRelease(&slot->mutex);
+
+ xmin_horizon = GetOldestSafeDecodingTransactionId(!need_full_snapshot);
+ ReplicationSlotsComputeRequiredXmin(true);
+
+ LWLockRelease(ProcArrayLock);
+ }
+
ctx = StartupDecodingContext(output_plugin_options,
- start_lsn, InvalidTransactionId, false,
+ start_lsn, xmin_horizon, need_full_snapshot,
fast_forward, xl_routine, prepare_write,
do_write, update_progress);
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index 5c23178570..e4b80f2764 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -208,6 +208,7 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
ctx = CreateDecodingContext(InvalidXLogRecPtr,
options,
false,
+ false,
XL_ROUTINE(.page_read = read_local_xlog_page,
.segment_open = wal_segment_open,
.segment_close = wal_segment_close),
diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index 7ba33d6672..f70b2e087a 100644
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -65,6 +65,7 @@ Node *replication_parse_result;
%token K_CREATE_REPLICATION_SLOT
%token K_DROP_REPLICATION_SLOT
%token K_TIMELINE_HISTORY
+%token K_REPLICATION_SLOT_SNAPSHOT
%token K_WAIT
%token K_TIMELINE
%token K_PHYSICAL
@@ -80,7 +81,7 @@ Node *replication_parse_result;
%type <node> command
%type <node> base_backup start_replication start_logical_replication
create_replication_slot drop_replication_slot identify_system
- read_replication_slot timeline_history show
+ read_replication_slot timeline_history show replication_slot_snapshot
%type <list> generic_option_list
%type <defelt> generic_option
%type <uintval> opt_timeline
@@ -114,6 +115,7 @@ command:
| read_replication_slot
| timeline_history
| show
+ | replication_slot_snapshot
;
/*
@@ -307,6 +309,19 @@ timeline_history:
}
;
+/*
+ * REPLICATION_SLOT_SNAPSHOT %s options
+ */
+replication_slot_snapshot:
+ K_REPLICATION_SLOT_SNAPSHOT var_name plugin_options
+ {
+ ReplicationSlotSnapshotCmd *n = makeNode(ReplicationSlotSnapshotCmd);
+ n->slotname = $2;
+ n->options = $3;
+ $$ = (Node *) n;
+ }
+ ;
+
opt_physical:
K_PHYSICAL
| /* EMPTY */
@@ -400,6 +415,7 @@ ident_or_keyword:
| K_CREATE_REPLICATION_SLOT { $$ = "create_replication_slot"; }
| K_DROP_REPLICATION_SLOT { $$ = "drop_replication_slot"; }
| K_TIMELINE_HISTORY { $$ = "timeline_history"; }
+ | K_REPLICATION_SLOT_SNAPSHOT { $$ = "replication_slot_snapshot"; }
| K_WAIT { $$ = "wait"; }
| K_TIMELINE { $$ = "timeline"; }
| K_PHYSICAL { $$ = "physical"; }
diff --git a/src/backend/replication/repl_scanner.l b/src/backend/replication/repl_scanner.l
index 72ef12225e..e33030da61 100644
--- a/src/backend/replication/repl_scanner.l
+++ b/src/backend/replication/repl_scanner.l
@@ -126,6 +126,7 @@ START_REPLICATION { return K_START_REPLICATION; }
CREATE_REPLICATION_SLOT { return K_CREATE_REPLICATION_SLOT; }
DROP_REPLICATION_SLOT { return K_DROP_REPLICATION_SLOT; }
TIMELINE_HISTORY { return K_TIMELINE_HISTORY; }
+REPLICATION_SLOT_SNAPSHOT { return K_REPLICATION_SLOT_SNAPSHOT; }
PHYSICAL { return K_PHYSICAL; }
RESERVE_WAL { return K_RESERVE_WAL; }
LOGICAL { return K_LOGICAL; }
@@ -303,6 +304,7 @@ replication_scanner_is_replication_command(void)
case K_DROP_REPLICATION_SLOT:
case K_READ_REPLICATION_SLOT:
case K_TIMELINE_HISTORY:
+ case K_REPLICATION_SLOT_SNAPSHOT:
case K_SHOW:
/* Yes; push back the first token so we can parse later. */
repl_pushed_back_token = first_token;
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 79d7f1372d..13d099fe06 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -478,6 +478,7 @@ pg_logical_replication_slot_advance(XLogRecPtr moveto)
ctx = CreateDecodingContext(InvalidXLogRecPtr,
NIL,
true, /* fast_forward */
+ false,
XL_ROUTINE(.page_read = read_local_xlog_page,
.segment_open = wal_segment_open,
.segment_close = wal_segment_close),
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index c11bb3716f..f9a22c8b16 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -238,6 +238,7 @@ static void CreateReplicationSlot(CreateReplicationSlotCmd *cmd);
static void DropReplicationSlot(DropReplicationSlotCmd *cmd);
static void StartReplication(StartReplicationCmd *cmd);
static void StartLogicalReplication(StartReplicationCmd *cmd);
+static void ReplicationSlotSnapshot(ReplicationSlotSnapshotCmd *cmd);
static void ProcessStandbyMessage(void);
static void ProcessStandbyReplyMessage(void);
static void ProcessStandbyHSFeedbackMessage(void);
@@ -1280,7 +1281,7 @@ StartLogicalReplication(StartReplicationCmd *cmd)
* are reported early.
*/
logical_decoding_ctx =
- CreateDecodingContext(cmd->startpoint, cmd->options, false,
+ CreateDecodingContext(cmd->startpoint, cmd->options, false, false,
XL_ROUTINE(.page_read = logical_read_xlog_page,
.segment_open = WalSndSegmentOpen,
.segment_close = wal_segment_close),
@@ -1332,6 +1333,91 @@ StartLogicalReplication(StartReplicationCmd *cmd)
EndCommand(&qc, DestRemote, false);
}
+/*
+ * Create a snapshot from an existing replication slot.
+ */
+static void
+ReplicationSlotSnapshot(ReplicationSlotSnapshotCmd *cmd)
+{
+ Snapshot snap;
+ LogicalDecodingContext *ctx;
+ char xloc[MAXFNAMELEN];
+ DestReceiver *dest;
+ TupOutputState *tstate;
+ TupleDesc tupdesc;
+ Datum values[1];
+ bool nulls[1] = {0};
+
+ Assert(!MyReplicationSlot);
+
+ if (!IsTransactionBlock())
+ ereport(ERROR,
+ (errmsg("%s must be called inside a transaction",
+ "REPLICATION_SLOT_SNAPSHOT ...")));
+
+ if (XactIsoLevel != XACT_REPEATABLE_READ)
+ ereport(ERROR,
+ (errmsg("%s must be called in REPEATABLE READ isolation mode transaction",
+ "REPLICATION_SLOT_SNAPSHOT ...")));
+
+ if (FirstSnapshotSet)
+ ereport(ERROR,
+ (errmsg("%s must be called before any query",
+ "REPLICATION_SLOT_SNAPSHOT ...")));
+
+ if (IsSubTransaction())
+ ereport(ERROR,
+ (errmsg("%s must not be called in a subtransaction",
+ "REPLICATION_SLOT_SNAPSHOT ...")));
+
+ ReplicationSlotAcquire(cmd->slotname, false);
+
+ ctx = CreateDecodingContext(MyReplicationSlot->data.restart_lsn,
+ cmd->options,
+ false,
+ true,
+ XL_ROUTINE(.page_read = logical_read_xlog_page,
+ .segment_open = WalSndSegmentOpen,
+ .segment_close = wal_segment_close),
+ WalSndPrepareWrite, WalSndWriteData,
+ WalSndUpdateProgress);
+
+ /*
+ * Signal that we don't need the timeout mechanism. We're just creating
+ * the replication slot and don't yet accept feedback messages or send
+ * keepalives. As we possibly need to wait for further WAL the walsender
+ * would otherwise possibly be killed too soon.
+ */
+ last_reply_timestamp = 0;
+
+ /* build initial snapshot, might take a while */
+ DecodingContextFindStartpoint(ctx);
+
+ snap = SnapBuildInitialSnapshot(ctx->snapshot_builder);
+ RestoreTransactionSnapshot(snap, MyProc);
+
+ /* Don't need the decoding context anymore */
+ FreeDecodingContext(ctx);
+
+ /* Create a tuple to send consisten WAL location */
+ snprintf(xloc, sizeof(xloc), "%X/%X",
+ LSN_FORMAT_ARGS(MyReplicationSlot->data.confirmed_flush));
+
+ dest = CreateDestReceiver(DestRemoteSimple);
+ tupdesc = CreateTemplateTupleDesc(1);
+ TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 1, "consistent_point",
+ TEXTOID, -1, 0);
+ tstate = begin_tup_output_tupdesc(dest, tupdesc, &TTSOpsVirtual);
+
+ /* consistent wal location */
+ values[0] = CStringGetTextDatum(xloc);
+
+ do_tup_output(tstate, values, nulls);
+ end_tup_output(tstate);
+
+ ReplicationSlotRelease();
+}
+
/*
* LogicalDecodingContext 'prepare_write' callback.
*
@@ -1860,6 +1946,15 @@ exec_replication_command(const char *cmd_string)
}
break;
+ case T_ReplicationSlotSnapshotCmd:
+ {
+ cmdtag = "REPLICATION_SLOT_SNAPSHOT";
+ set_ps_display(cmdtag);
+ ReplicationSlotSnapshot((ReplicationSlotSnapshotCmd *) cmd_node);
+ EndReplicationCommand(cmdtag);
+ break;
+ }
+
default:
elog(ERROR, "unrecognized replication command node tag: %u",
cmd_node->type);
diff --git a/src/include/nodes/replnodes.h b/src/include/nodes/replnodes.h
index 8ae9c517ff..94aa7d2253 100644
--- a/src/include/nodes/replnodes.h
+++ b/src/include/nodes/replnodes.h
@@ -108,4 +108,15 @@ typedef struct TimeLineHistoryCmd
TimeLineID timeline;
} TimeLineHistoryCmd;
+/* ----------------------
+ * REPLICATION_SLOT_SNAPSHOT command
+ * ----------------------
+ */
+typedef struct ReplicationSlotSnapshotCmd
+{
+ NodeTag type;
+ char *slotname;
+ List *options;
+} ReplicationSlotSnapshotCmd;
+
#endif /* REPLNODES_H */
diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h
index 4ad019e25a..3875b32f9d 100644
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -125,6 +125,7 @@ extern LogicalDecodingContext *CreateInitDecodingContext(const char *plugin,
extern LogicalDecodingContext *CreateDecodingContext(XLogRecPtr start_lsn,
List *output_plugin_options,
bool fast_forward,
+ bool need_full_snapshot,
XLogReaderRoutine *xl_routine,
LogicalOutputPluginWriterPrepareWrite prepare_write,
LogicalOutputPluginWriterWrite do_write,
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 9339f29303..274a162b48 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -384,6 +384,16 @@ typedef WalRcvExecResult *(*walrcv_exec_fn) (WalReceiverConn *conn,
*/
typedef void (*walrcv_disconnect_fn) (WalReceiverConn *conn);
+/*
+ * walrcv_slot_snapshot_fn
+ *
+ * Create a snapshot by an existing replication slot
+ */
+typedef void (*walrcv_slot_snapshot_fn) (WalReceiverConn *conn,
+ char *slotname,
+ const WalRcvStreamOptions *options,
+ XLogRecPtr *lsn);
+
typedef struct WalReceiverFunctionsType
{
walrcv_connect_fn walrcv_connect;
@@ -401,6 +411,7 @@ typedef struct WalReceiverFunctionsType
walrcv_get_backend_pid_fn walrcv_get_backend_pid;
walrcv_exec_fn walrcv_exec;
walrcv_disconnect_fn walrcv_disconnect;
+ walrcv_slot_snapshot_fn walrcv_slot_snapshot;
} WalReceiverFunctionsType;
extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
@@ -435,6 +446,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
WalReceiverFunctions->walrcv_exec(conn, exec, nRetTypes, retTypes)
#define walrcv_disconnect(conn) \
WalReceiverFunctions->walrcv_disconnect(conn)
+#define walrcv_slot_snapshot(conn, slotname, options, lsn) \
+ WalReceiverFunctions->walrcv_slot_snapshot(conn, slotname, options, lsn)
static inline void
walrcv_clear_result(WalRcvExecResult *walres)
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 60c71d05fe..6b90a4a693 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2315,6 +2315,7 @@ ReplicationSlotCtlData
ReplicationSlotOnDisk
ReplicationSlotPersistency
ReplicationSlotPersistentData
+ReplicationSlotSnapshotCmd
ReplicationState
ReplicationStateCtl
ReplicationStateOnDisk
--
2.25.1
[application/octet-stream] v5-0002-Reuse-Logical-Replication-Background-worker.patch (73.3K, ../CAGPVpCQdZ_oj-QFcTOhTrUTs-NCKrrZ=ZNCNPR1qe27rXV-iYw@mail.gmail.com/4-v5-0002-Reuse-Logical-Replication-Background-worker.patch)
download | inline diff:
From ce74b9a18a2a425ed39533f435fefaa2431285c2 Mon Sep 17 00:00:00 2001
From: Melih Mutlu <[email protected]>
Date: Thu, 2 Jun 2022 17:39:37 +0300
Subject: [PATCH 2/2] Reuse Logical Replication Background worker
This commit allows tablesync workers to move to another table that needs synchronization,
when they're done with the current table in tablesync phase of Logical Replication.
Before this commit, tablesync workers were capable of syncing only one
relation. A new worker, replication slot and origin were needed for each
relation in the subscription.
Now, tablesync workers are not only limited with one relation and can move to another relation and reuse existing
replication slots and origins
This reduces the overhead of launching/killing a new background worker for each relation.
By reusing tablesync workers, replication slots and origins created for tablesync can be reused as well.
Removing the burden of creating/dropping replication slot/origin improves tablesync speed significantly especially for empty or small tables.
Discussion: http://postgr.es/m/CAGPVpCTq=rUDd4JUdaRc1XUWf4BrH2gdSNf3rtOMUGj9rPpfzQ@mail.gmail.com
---
doc/src/sgml/catalogs.sgml | 29 ++
src/backend/catalog/pg_subscription.c | 284 ++++++++++++-
src/backend/commands/subscriptioncmds.c | 238 +++++++----
src/backend/replication/logical/launcher.c | 9 +-
src/backend/replication/logical/tablesync.c | 434 +++++++++++++++-----
src/backend/replication/logical/worker.c | 370 +++++++++++------
src/include/catalog/pg_subscription.h | 6 +
src/include/catalog/pg_subscription_rel.h | 15 +-
src/include/replication/slot.h | 3 +-
src/include/replication/worker_internal.h | 32 +-
src/test/regress/expected/misc_sanity.out | 30 +-
11 files changed, 1095 insertions(+), 355 deletions(-)
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 9316b811ac..7d55db6354 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -7997,6 +7997,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>sublastusedid</structfield> <type>int8</type>
+ </para>
+ <para>
+ The last used ID for tablesync workers. This ID is used to
+ create replication slots. The last used ID needs to be stored
+ to make logical replication can safely proceed after any interruption.
+ </para></entry>
+ </row>
</tbody>
</tgroup>
</table>
@@ -8081,6 +8092,24 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
otherwise null
</para></entry>
</row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>srrelslotname</structfield> <type>text</type>
+ </para>
+ <para>
+ Replication slot name that is used for synchronization of relation
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>srreloriginname</structfield> <type>text</type>
+ </para>
+ <para>
+ Origin name that is used for tracking synchronization of relation
+ </para></entry>
+ </row>
</tbody>
</tgroup>
</table>
diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index a506fc3ec8..1f637df242 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -114,6 +114,14 @@ GetSubscription(Oid subid, bool missing_ok)
Assert(!isnull);
sub->origin = TextDatumGetCString(datum);
+ /* Get last used id */
+ datum = SysCacheGetAttr(SUBSCRIPTIONOID,
+ tup,
+ Anum_pg_subscription_sublastusedid,
+ &isnull);
+ Assert(!isnull);
+ sub->lastusedid = DatumGetInt64(datum);
+
ReleaseSysCache(tup);
return sub;
@@ -205,6 +213,44 @@ DisableSubscription(Oid subid)
table_close(rel, NoLock);
}
+/*
+ * Update the last used replication slot ID for the given subscription.
+ */
+void
+UpdateSubscriptionLastSlotId(Oid subid, int64 lastusedid)
+{
+ Relation rel;
+ bool nulls[Natts_pg_subscription];
+ bool replaces[Natts_pg_subscription];
+ Datum values[Natts_pg_subscription];
+ HeapTuple tup;
+
+ /* Look up the subscription in the catalog */
+ rel = table_open(SubscriptionRelationId, RowExclusiveLock);
+ tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
+
+ if (!HeapTupleIsValid(tup))
+ elog(ERROR, "cache lookup failed for subscription %u", subid);
+
+ LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
+
+ /* Form a new tuple. */
+ memset(values, 0, sizeof(values));
+ memset(nulls, false, sizeof(nulls));
+ memset(replaces, false, sizeof(replaces));
+
+ replaces[Anum_pg_subscription_sublastusedid - 1] = true;
+ values[Anum_pg_subscription_sublastusedid- 1] = Int64GetDatum(lastusedid);
+
+ /* Update the catalog */
+ tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
+ replaces);
+ CatalogTupleUpdate(rel, &tup->t_self, tup);
+ heap_freetuple(tup);
+
+ table_close(rel, NoLock);
+}
+
/*
* Convert text array to list of strings.
*
@@ -234,7 +280,7 @@ textarray_to_stringlist(ArrayType *textarray)
*/
void
AddSubscriptionRelState(Oid subid, Oid relid, char state,
- XLogRecPtr sublsn)
+ XLogRecPtr sublsn, char *relslotname, char *reloriginname)
{
Relation rel;
HeapTuple tup;
@@ -263,6 +309,14 @@ AddSubscriptionRelState(Oid subid, Oid relid, char state,
values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
else
nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
+ if (relslotname)
+ values[Anum_pg_subscription_rel_srrelslotname - 1] = CStringGetTextDatum(relslotname);
+ else
+ nulls[Anum_pg_subscription_rel_srrelslotname - 1] = true;
+ if (reloriginname)
+ values[Anum_pg_subscription_rel_srreloriginname - 1] = CStringGetTextDatum(reloriginname);
+ else
+ nulls[Anum_pg_subscription_rel_srreloriginname - 1] = true;
tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
@@ -275,6 +329,58 @@ AddSubscriptionRelState(Oid subid, Oid relid, char state,
table_close(rel, NoLock);
}
+/*
+ * Internal function to modify columns for relation state update
+ */
+static void
+UpdateSubscriptionRelState_internal(Datum *values,
+ bool *nulls,
+ bool *replaces,
+ char state,
+ XLogRecPtr sublsn)
+{
+ replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
+ values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
+
+ replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
+ if (sublsn != InvalidXLogRecPtr)
+ values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
+ else
+ nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
+}
+
+/*
+ * Internal function to modify columns for replication slot update
+ */
+static void
+UpdateSubscriptionRelReplicationSlot_internal(Datum *values,
+ bool *nulls,
+ bool *replaces,
+ char *relslotname)
+{
+ replaces[Anum_pg_subscription_rel_srrelslotname - 1] = true;
+ if (relslotname)
+ values[Anum_pg_subscription_rel_srrelslotname - 1] = CStringGetTextDatum(relslotname);
+ else
+ nulls[Anum_pg_subscription_rel_srrelslotname - 1] = true;
+}
+
+/*
+ * Internal function to modify columns for replication origin update
+ */
+static void
+UpdateSubscriptionRelOrigin_internal(Datum *values,
+ bool *nulls,
+ bool *replaces,
+ char *reloriginname)
+{
+ replaces[Anum_pg_subscription_rel_srreloriginname - 1] = true;
+ if (reloriginname)
+ values[Anum_pg_subscription_rel_srreloriginname - 1] = CStringGetTextDatum(reloriginname);
+ else
+ nulls[Anum_pg_subscription_rel_srreloriginname - 1] = true;
+}
+
/*
* Update the state of a subscription table.
*/
@@ -305,14 +411,48 @@ UpdateSubscriptionRelState(Oid subid, Oid relid, char state,
memset(nulls, false, sizeof(nulls));
memset(replaces, false, sizeof(replaces));
- replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
- values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
+ UpdateSubscriptionRelState_internal(values, nulls, replaces, state, sublsn);
- replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
- if (sublsn != InvalidXLogRecPtr)
- values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
- else
- nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
+ tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
+ replaces);
+
+ /* Update the catalog. */
+ CatalogTupleUpdate(rel, &tup->t_self, tup);
+
+ /* Cleanup. */
+ table_close(rel, NoLock);
+}
+
+/*
+ * Update the replication slot name of a subscription table.
+ */
+void
+UpdateSubscriptionRelReplicationSlot(Oid subid, Oid relid, char *relslotname)
+{
+ Relation rel;
+ HeapTuple tup;
+ bool nulls[Natts_pg_subscription_rel];
+ Datum values[Natts_pg_subscription_rel];
+ bool replaces[Natts_pg_subscription_rel];
+
+ LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
+
+ rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
+
+ /* Try finding existing mapping. */
+ tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
+ ObjectIdGetDatum(relid),
+ ObjectIdGetDatum(subid));
+ if (!HeapTupleIsValid(tup))
+ elog(ERROR, "subscription table %u in subscription %u does not exist",
+ relid, subid);
+
+ /* Update the tuple. */
+ memset(values, 0, sizeof(values));
+ memset(nulls, false, sizeof(nulls));
+ memset(replaces, false, sizeof(replaces));
+
+ UpdateSubscriptionRelReplicationSlot_internal(values, nulls, replaces, relslotname);
tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
replaces);
@@ -324,6 +464,134 @@ UpdateSubscriptionRelState(Oid subid, Oid relid, char state,
table_close(rel, NoLock);
}
+/*
+ * Update replication slot name, origin name and state of
+ * a subscription table in one transaction.
+ */
+void
+UpdateSubscriptionRel(Oid subid,
+ Oid relid,
+ char state,
+ XLogRecPtr sublsn,
+ char *relslotname,
+ char *reloriginname)
+{
+ Relation rel;
+ HeapTuple tup;
+ bool nulls[Natts_pg_subscription_rel];
+ Datum values[Natts_pg_subscription_rel];
+ bool replaces[Natts_pg_subscription_rel];
+
+ LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
+
+ rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
+
+ /* Try finding existing mapping. */
+ tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
+ ObjectIdGetDatum(relid),
+ ObjectIdGetDatum(subid));
+ if (!HeapTupleIsValid(tup))
+ elog(ERROR, "subscription table %u in subscription %u does not exist",
+ relid, subid);
+
+ /* Update the tuple. */
+ memset(values, 0, sizeof(values));
+ memset(nulls, false, sizeof(nulls));
+ memset(replaces, false, sizeof(replaces));
+
+ UpdateSubscriptionRelState_internal(values, nulls, replaces, state, sublsn);
+ UpdateSubscriptionRelReplicationSlot_internal(values, nulls, replaces, relslotname);
+ UpdateSubscriptionRelOrigin_internal(values, nulls, replaces, reloriginname);
+
+ tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
+ replaces);
+
+ /* Update the catalog. */
+ CatalogTupleUpdate(rel, &tup->t_self, tup);
+
+ /* Cleanup. */
+ table_close(rel, NoLock);
+}
+
+/*
+ * Get origin name of subscription table.
+ *
+ * Returns null if the subscription table does not have a origin.
+ */
+void
+GetSubscriptionRelOrigin(Oid subid, Oid relid, char *reloriginname, bool *isnull)
+{
+ HeapTuple tup;
+ Relation rel;
+ Datum d;
+ char *originname;
+
+ rel = table_open(SubscriptionRelRelationId, AccessShareLock);
+
+ /* Try finding the mapping. */
+ tup = SearchSysCache2(SUBSCRIPTIONRELMAP,
+ ObjectIdGetDatum(relid),
+ ObjectIdGetDatum(subid));
+
+ if (!HeapTupleIsValid(tup))
+ {
+ table_close(rel, AccessShareLock);
+ }
+
+ d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
+ Anum_pg_subscription_rel_srreloriginname, isnull);
+ if (!*isnull)
+ {
+ originname = TextDatumGetCString(d);
+ memcpy(reloriginname, originname, NAMEDATALEN);
+ }
+
+ /* Cleanup */
+ ReleaseSysCache(tup);
+
+ table_close(rel, AccessShareLock);
+}
+
+/*
+ * Get replication slot name of subscription table.
+ *
+ * Returns null if the subscription table does not have a replication slot.
+ */
+void
+GetSubscriptionRelReplicationSlot(Oid subid, Oid relid, char *slotname)
+{
+ HeapTuple tup;
+ Relation rel;
+ Datum d;
+ char *relrepslot;
+ bool isnull;
+
+ rel = table_open(SubscriptionRelRelationId, AccessShareLock);
+
+ /* Try finding the mapping. */
+ tup = SearchSysCache2(SUBSCRIPTIONRELMAP,
+ ObjectIdGetDatum(relid),
+ ObjectIdGetDatum(subid));
+
+ if (!HeapTupleIsValid(tup))
+ {
+ table_close(rel, AccessShareLock);
+ }
+
+ d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
+ Anum_pg_subscription_rel_srrelslotname, &isnull);
+ if (!isnull)
+ {
+ relrepslot = TextDatumGetCString(d);
+ memcpy(slotname, relrepslot, NAMEDATALEN);
+ }
+
+ /* Cleanup */
+ ReleaseSysCache(tup);
+
+ table_close(rel, AccessShareLock);
+}
+
/*
* Get state of subscription table.
*
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index d673557ea4..ccdd844201 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -648,6 +648,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
publicationListToArray(publications);
values[Anum_pg_subscription_suborigin - 1] =
CStringGetTextDatum(opts.origin);
+ values[Anum_pg_subscription_sublastusedid - 1] = Int64GetDatum(1);
tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
@@ -657,7 +658,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
recordDependencyOnOwner(SubscriptionRelationId, subid, owner);
- ReplicationOriginNameForLogicalRep(subid, InvalidOid, originname, sizeof(originname));
+ ReplicationOriginNameForLogicalRep(subid, originname, sizeof(originname), false);
replorigin_create(originname);
/*
@@ -708,7 +709,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
rv->schemaname, rv->relname);
AddSubscriptionRelState(subid, relid, table_state,
- InvalidXLogRecPtr);
+ InvalidXLogRecPtr, NULL, NULL);
}
/*
@@ -798,6 +799,8 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
} SubRemoveRels;
SubRemoveRels *sub_remove_rels;
WalReceiverConn *wrconn;
+ List *sub_remove_slots = NIL;
+ LogicalRepWorker *worker;
/* Load the library providing us libpq calls. */
load_file("libpqwalreceiver", false);
@@ -875,7 +878,7 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
{
AddSubscriptionRelState(sub->oid, relid,
copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY,
- InvalidXLogRecPtr);
+ InvalidXLogRecPtr, NULL, NULL);
ereport(DEBUG1,
(errmsg_internal("table \"%s.%s\" added to subscription \"%s\"",
rv->schemaname, rv->relname, sub->name)));
@@ -899,6 +902,7 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
{
char state;
XLogRecPtr statelsn;
+ char slotname[NAMEDATALEN] = {0};
/*
* Lock pg_subscription_rel with AccessExclusiveLock to
@@ -925,7 +929,29 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
RemoveSubscriptionRel(sub->oid, relid);
- logicalrep_worker_stop(sub->oid, relid);
+ /*
+ * Find the logical replication sync worker if exists store
+ * the slot number for dropping associated replication slots
+ * later.
+ */
+ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+ worker = logicalrep_worker_find(sub->oid, relid, false);
+ if (worker)
+ {
+ logicalrep_worker_stop(sub->oid, relid);
+ sub_remove_slots = lappend(sub_remove_slots, &worker->slot_name);
+ }
+ else
+ {
+ /*
+ * Sync of this relation might be failed in an earlier
+ * attempt, but the replication slot might still exist.
+ */
+ GetSubscriptionRelReplicationSlot(sub->oid, relid, slotname);
+ if (strlen(slotname) > 0)
+ sub_remove_slots = lappend(sub_remove_slots, slotname);
+ }
+ LWLockRelease(LogicalRepWorkerLock);
/*
* For READY state, we would have already dropped the
@@ -945,8 +971,8 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
* origin and by this time the origin might be already
* removed. For these reasons, passing missing_ok = true.
*/
- ReplicationOriginNameForLogicalRep(sub->oid, relid, originname,
- sizeof(originname));
+ ReplicationOriginNameForLogicalRep(sub->oid, originname,
+ sizeof(originname), true);
replorigin_drop_by_name(originname, true, false);
}
@@ -959,31 +985,24 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
}
/*
- * Drop the tablesync slots associated with removed tables. This has
- * to be at the end because otherwise if there is an error while doing
- * the database operations we won't be able to rollback dropped slots.
+ * Drop the replication slots associated with tablesync workers for
+ * removed tables. This has to be at the end because otherwise if
+ * there is an error while doing the database operations we won't be
+ * able to rollback dropped slots.
*/
- for (off = 0; off < remove_rel_len; off++)
+ foreach(lc, sub_remove_slots)
{
- if (sub_remove_rels[off].state != SUBREL_STATE_READY &&
- sub_remove_rels[off].state != SUBREL_STATE_SYNCDONE)
- {
- char syncslotname[NAMEDATALEN] = {0};
+ char syncslotname[NAMEDATALEN] = {0};
- /*
- * For READY/SYNCDONE states we know the tablesync slot has
- * already been dropped by the tablesync worker.
- *
- * For other states, there is no certainty, maybe the slot
- * does not exist yet. Also, if we fail after removing some of
- * the slots, next time, it will again try to drop already
- * dropped slots and fail. For these reasons, we allow
- * missing_ok = true for the drop.
- */
- ReplicationSlotNameForTablesync(sub->oid, sub_remove_rels[off].relid,
- syncslotname, sizeof(syncslotname));
- ReplicationSlotDropAtPubNode(wrconn, syncslotname, true);
- }
+ memcpy(syncslotname, lfirst(lc), sizeof(NAMEDATALEN));
+
+ /*
+ * There is no certainty, maybe the slot does not exist yet. Also,
+ * if we fail after removing some of the slots, next time, it will
+ * again try to drop already dropped slots and fail. For these
+ * reasons, we allow missing_ok = true for the drop.
+ */
+ ReplicationSlotDropAtPubNode(wrconn, syncslotname, true);
}
}
PG_FINALLY();
@@ -1319,8 +1338,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
char originname[NAMEDATALEN];
XLogRecPtr remote_lsn;
- ReplicationOriginNameForLogicalRep(subid, InvalidOid,
- originname, sizeof(originname));
+ ReplicationOriginNameForLogicalRep(subid, originname,
+ sizeof(originname), false);
originid = replorigin_by_name(originname, false);
remote_lsn = replorigin_get_progress(originid, false);
@@ -1380,6 +1399,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
char *subname;
char *conninfo;
char *slotname;
+ int64 lastusedid;
List *subworkers;
ListCell *lc;
char originname[NAMEDATALEN];
@@ -1451,6 +1471,14 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
else
slotname = NULL;
+ /* Get the last used identifier by the subscription */
+ datum = SysCacheGetAttr(SUBSCRIPTIONOID, tup,
+ Anum_pg_subscription_sublastusedid, &isnull);
+ if (!isnull)
+ lastusedid = DatumGetInt64(datum);
+ else
+ lastusedid = 0;
+
/*
* Since dropping a replication slot is not transactional, the replication
* slot stays dropped even if the transaction rolls back. So we cannot
@@ -1500,37 +1528,29 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
}
list_free(subworkers);
+ rstates = GetSubscriptionRelations(subid, true);
+
/*
- * Cleanup of tablesync replication origins.
- *
- * Any READY-state relations would already have dealt with clean-ups.
+ * Cleanup of tablesync replication origins associated with the
+ * subscription, if exists. Try to drop origins by creating all origin
+ * names created for this subscription.
*
* Note that the state can't change because we have already stopped both
* the apply and tablesync workers and they can't restart because of
* exclusive lock on the subscription.
+ *
+ * XXX: This can be handled better instead of looping through all possible
*/
- rstates = GetSubscriptionRelations(subid, true);
- foreach(lc, rstates)
+ for (int64 i = 1; i <= lastusedid; i++)
{
- SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
- Oid relid = rstate->relid;
-
- /* Only cleanup resources of tablesync workers */
- if (!OidIsValid(relid))
- continue;
+ char originname_to_drop[NAMEDATALEN] = {0};
- /*
- * Drop the tablesync's origin tracking if exists.
- *
- * It is possible that the origin is not yet created for tablesync
- * worker so passing missing_ok = true. This can happen for the states
- * before SUBREL_STATE_FINISHEDCOPY.
- */
- ReplicationOriginNameForLogicalRep(subid, relid, originname,
- sizeof(originname));
- replorigin_drop_by_name(originname, true, false);
+ snprintf(originname_to_drop, sizeof(originname_to_drop), "pg_%u_%ld", subid, i);
+ /* missin_ok = true, since the origin might be already dropped. */
+ replorigin_drop_by_name(originname_to_drop, true, false);
}
+
/* Clean up dependencies */
deleteSharedDependencyRecordsFor(SubscriptionRelationId, subid, 0);
@@ -1538,7 +1558,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
RemoveSubscriptionRel(subid, InvalidOid);
/* Remove the origin tracking if exists. */
- ReplicationOriginNameForLogicalRep(subid, InvalidOid, originname, sizeof(originname));
+ ReplicationOriginNameForLogicalRep(subid, originname, sizeof(originname), false);
replorigin_drop_by_name(originname, true, false);
/*
@@ -1582,39 +1602,17 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
PG_TRY();
{
- foreach(lc, rstates)
- {
- SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
- Oid relid = rstate->relid;
+ List *slots = NULL;
- /* Only cleanup resources of tablesync workers */
- if (!OidIsValid(relid))
- continue;
- /*
- * Drop the tablesync slots associated with removed tables.
- *
- * For SYNCDONE/READY states, the tablesync slot is known to have
- * already been dropped by the tablesync worker.
- *
- * For other states, there is no certainty, maybe the slot does
- * not exist yet. Also, if we fail after removing some of the
- * slots, next time, it will again try to drop already dropped
- * slots and fail. For these reasons, we allow missing_ok = true
- * for the drop.
- */
- if (rstate->state != SUBREL_STATE_SYNCDONE)
- {
- char syncslotname[NAMEDATALEN] = {0};
+ slots = GetReplicationSlotNamesBySubId(wrconn, subid, true);
+ foreach(lc, slots)
+ {
+ char *syncslotname = (char *) lfirst(lc);
- ReplicationSlotNameForTablesync(subid, relid, syncslotname,
- sizeof(syncslotname));
- ReplicationSlotDropAtPubNode(wrconn, syncslotname, true);
- }
+ ReplicationSlotDropAtPubNode(wrconn, syncslotname, true);
}
- list_free(rstates);
-
/*
* If there is a slot associated with the subscription, then drop the
* replication slot at the publisher.
@@ -1637,6 +1635,71 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
table_close(rel, NoLock);
}
+/*
+ * GetReplicationSlotNamesBySubId
+ *
+ * Get the replication slot names associated with the subscription.
+ */
+List *
+GetReplicationSlotNamesBySubId(WalReceiverConn *wrconn, Oid subid, bool missing_ok)
+{
+ StringInfoData cmd;
+ TupleTableSlot *slot;
+ Oid tableRow[1] = {NAMEOID};
+ List *tablelist = NIL;
+
+ Assert(wrconn);
+
+ load_file("libpqwalreceiver", false);
+
+ initStringInfo(&cmd);
+ appendStringInfo(&cmd, "SELECT slot_name"
+ " FROM pg_replication_slots"
+ " WHERE slot_name LIKE 'pg_%i_sync_%%';",
+ subid);
+ PG_TRY();
+ {
+ WalRcvExecResult *res;
+
+ res = walrcv_exec(wrconn, cmd.data, 1, tableRow);
+
+ if (res->status != WALRCV_OK_TUPLES)
+ {
+ ereport(ERROR,
+ errmsg("not tuple returned."));
+ }
+
+ /* Process tables. */
+ slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
+ while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
+ {
+ char *repslotname;
+ char *slotattr;
+ bool isnull;
+
+ slotattr = NameStr(*DatumGetName(slot_getattr(slot, 1, &isnull)));
+ Assert(!isnull);
+
+ repslotname = palloc(sizeof(char) * strlen(slotattr) + 1);
+ memcpy(repslotname, slotattr, sizeof(char) * strlen(slotattr));
+ repslotname[strlen(slotattr)] = '\0';
+ tablelist = lappend(tablelist, repslotname);
+
+ ExecClearTuple(slot);
+ }
+ ExecDropSingleTupleTableSlot(slot);
+
+ walrcv_clear_result(res);
+ }
+ PG_FINALLY();
+ {
+ pfree(cmd.data);
+ }
+ PG_END_TRY();
+ \
+ return tablelist;
+}
+
/*
* Drop the replication slot at the publisher node using the replication
* connection.
@@ -1989,6 +2052,7 @@ static void
ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err)
{
ListCell *lc;
+ LogicalRepWorker *worker;
foreach(lc, rstates)
{
@@ -1999,18 +2063,20 @@ ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err)
if (!OidIsValid(relid))
continue;
+ /* Check if there is a sync worker for the relation */
+ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+ worker = logicalrep_worker_find(subid, relid, false);
+ LWLockRelease(LogicalRepWorkerLock);
+
/*
* Caller needs to ensure that relstate doesn't change underneath us.
* See DropSubscription where we get the relstates.
*/
- if (rstate->state != SUBREL_STATE_SYNCDONE)
+ if (worker &&
+ rstate->state != SUBREL_STATE_SYNCDONE)
{
- char syncslotname[NAMEDATALEN] = {0};
-
- ReplicationSlotNameForTablesync(subid, relid, syncslotname,
- sizeof(syncslotname));
elog(WARNING, "could not drop tablesync replication slot \"%s\"",
- syncslotname);
+ worker->slot_name);
}
}
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 73594c698a..5b5dead996 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -264,7 +264,7 @@ logicalrep_workers_find(Oid subid, bool only_running)
*/
void
logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
- Oid relid)
+ Oid relid, int64 slotid)
{
BackgroundWorker bgw;
BackgroundWorkerHandle *bgw_handle;
@@ -371,7 +371,11 @@ retry:
/* Prepare the worker slot. */
worker->launch_time = now;
worker->in_use = true;
+ worker->is_first_run = true;
worker->generation++;
+ worker->created_slot = false;
+ worker->rep_slot_id = slotid;
+ worker->slot_name = (char *) palloc(NAMEDATALEN);
worker->proc = NULL;
worker->dbid = dbid;
worker->userid = userid;
@@ -379,6 +383,7 @@ retry:
worker->relid = relid;
worker->relstate = SUBREL_STATE_UNKNOWN;
worker->relstate_lsn = InvalidXLogRecPtr;
+ worker->move_to_next_rel = false;
worker->stream_fileset = NULL;
worker->last_lsn = InvalidXLogRecPtr;
TIMESTAMP_NOBEGIN(worker->last_send_time);
@@ -869,7 +874,7 @@ ApplyLauncherMain(Datum main_arg)
wait_time = wal_retrieve_retry_interval;
logicalrep_worker_launch(sub->dbid, sub->oid, sub->name,
- sub->owner, InvalidOid);
+ sub->owner, InvalidOid, 0);
}
}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 94e813ac53..a6364d9433 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -126,12 +126,8 @@ static bool FetchTableStates(bool *started_tx);
static StringInfo copybuf = NULL;
-/*
- * Exit routine for synchronization worker.
- */
static void
-pg_attribute_noreturn()
-finish_sync_worker(void)
+clean_sync_worker(void)
{
/*
* Commit any outstanding transaction. This is the usual case, unless
@@ -143,18 +139,28 @@ finish_sync_worker(void)
pgstat_report_stat(true);
}
- /* And flush all writes. */
- XLogFlush(GetXLogWriteRecPtr());
-
- StartTransactionCommand();
- ereport(LOG,
- (errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has finished",
- MySubscription->name,
- get_rel_name(MyLogicalRepWorker->relid))));
- CommitTransactionCommand();
+ /*
+ * Disconnect from publisher. Otherwise reused sync workers causes
+ * exceeding max_wal_senders
+ */
+ walrcv_disconnect(LogRepWorkerWalRcvConn);
+ LogRepWorkerWalRcvConn = NULL;
/* Find the main apply worker and signal it. */
logicalrep_worker_wakeup(MyLogicalRepWorker->subid, InvalidOid);
+}
+
+/*
+ * Exit routine for synchronization worker.
+ */
+static void
+pg_attribute_noreturn()
+finish_sync_worker(void)
+{
+ clean_sync_worker();
+
+ /* And flush all writes. */
+ XLogFlush(GetXLogWriteRecPtr());
/* Stop gracefully */
proc_exit(0);
@@ -284,6 +290,10 @@ invalidate_syncing_table_states(Datum arg, int cacheid, uint32 hashvalue)
static void
process_syncing_tables_for_sync(XLogRecPtr current_lsn)
{
+ List *rstates;
+ SubscriptionRelState *rstate;
+ ListCell *lc;
+
SpinLockAcquire(&MyLogicalRepWorker->relmutex);
if (MyLogicalRepWorker->relstate == SUBREL_STATE_CATCHUP &&
@@ -292,6 +302,7 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn)
TimeLineID tli;
char syncslotname[NAMEDATALEN] = {0};
char originname[NAMEDATALEN] = {0};
+ bool is_streaming_ended = false;
MyLogicalRepWorker->relstate = SUBREL_STATE_SYNCDONE;
MyLogicalRepWorker->relstate_lsn = current_lsn;
@@ -308,40 +319,29 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn)
MyLogicalRepWorker->relid,
MyLogicalRepWorker->relstate,
MyLogicalRepWorker->relstate_lsn);
+ CommitTransactionCommand();
/*
- * End streaming so that LogRepWorkerWalRcvConn can be used to drop
- * the slot.
- */
- walrcv_endstreaming(LogRepWorkerWalRcvConn, &tli);
-
- /*
- * Cleanup the tablesync slot.
+ * Cleanup the tablesync slot. If the slot name used by this worker is
+ * different from the default slot name for the worker, this means the
+ * current table had started to being synchronized by another worker
+ * and replication slot. And this worker is reusing a replication slot
+ * from a previous attempt. We do not need that replication slot
+ * anymore.
*
* This has to be done after updating the state because otherwise if
* there is an error while doing the database operations we won't be
* able to rollback dropped slot.
*/
ReplicationSlotNameForTablesync(MyLogicalRepWorker->subid,
- MyLogicalRepWorker->relid,
+ MyLogicalRepWorker->rep_slot_id,
syncslotname,
sizeof(syncslotname));
/*
- * It is important to give an error if we are unable to drop the slot,
- * otherwise, it won't be dropped till the corresponding subscription
- * is dropped. So passing missing_ok = false.
- */
- ReplicationSlotDropAtPubNode(LogRepWorkerWalRcvConn, syncslotname, false);
-
- CommitTransactionCommand();
- pgstat_report_stat(false);
-
- /*
- * Start a new transaction to clean up the tablesync origin tracking.
- * This transaction will be ended within the finish_sync_worker().
- * Now, even, if we fail to remove this here, the apply worker will
- * ensure to clean it up afterward.
+ * We are safe to drop the replication trackin origin after this
+ * point. Now, even, if we fail to remove this here, the apply worker
+ * will ensure to clean it up afterward.
*
* We need to do this after the table state is set to SYNCDONE.
* Otherwise, if an error occurs while performing the database
@@ -350,34 +350,125 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn)
* have been cleared before restart. So, the restarted worker will use
* invalid replication progress state resulting in replay of
* transactions that have already been applied.
+ *
+ * Firstly reset the origin session to remove the ownership of the
+ * slot. This is needed to allow the origin to be dropped or reused
+ * later.
+ */
+ replorigin_session_reset();
+ replorigin_session_origin = InvalidRepOriginId;
+ replorigin_session_origin_lsn = InvalidXLogRecPtr;
+ replorigin_session_origin_timestamp = 0;
+
+ StartTransactionCommand();
+ if (MyLogicalRepWorker->slot_name && strcmp(syncslotname, MyLogicalRepWorker->slot_name) != 0)
+ {
+ /*
+ * End streaming so that LogRepWorkerWalRcvConn can be used to
+ * drop the slot.
+ */
+ walrcv_endstreaming(LogRepWorkerWalRcvConn, &tli);
+ is_streaming_ended = true;
+ ReplicationSlotDropAtPubNode(LogRepWorkerWalRcvConn, MyLogicalRepWorker->slot_name, false);
+
+ ReplicationOriginNameForLogicalRep(MyLogicalRepWorker->subid,
+ originname,
+ sizeof(originname),
+ true);
+
+ /* Drop replication origin */
+ replorigin_drop_by_name(originname, true, false);
+ }
+
+ /*
+ * We are safe to remove persisted replication slot and origin data,
+ * since it's already in SYNCDONE state. They will not be needed
+ * anymore.
*/
+ UpdateSubscriptionRel(MyLogicalRepWorker->subid,
+ MyLogicalRepWorker->relid,
+ MyLogicalRepWorker->relstate,
+ MyLogicalRepWorker->relstate_lsn,
+ NULL,
+ NULL);
+
+ ereport(LOG,
+ (errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has finished",
+ MySubscription->name,
+ get_rel_name(MyLogicalRepWorker->relid))));
+
+ CommitTransactionCommand();
+ pgstat_report_stat(false);
+
StartTransactionCommand();
+ /*
+ * This should return the default origin name for the worker. Even if
+ * the worker used a different origin for this table, it should be
+ * dropped and removed from the catalog so far.
+ */
ReplicationOriginNameForLogicalRep(MyLogicalRepWorker->subid,
- MyLogicalRepWorker->relid,
originname,
- sizeof(originname));
+ sizeof(originname),
+ true);
/*
- * Resetting the origin session removes the ownership of the slot.
- * This is needed to allow the origin to be dropped.
+ * Check if any table whose relation state is still INIT. If a table
+ * in INIT state is found, the worker will not be finished, it will be
+ * reused instead.
*/
- replorigin_session_reset();
- replorigin_session_origin = InvalidRepOriginId;
- replorigin_session_origin_lsn = InvalidXLogRecPtr;
- replorigin_session_origin_timestamp = 0;
+ rstates = GetSubscriptionRelations(MySubscription->oid, true);
- /*
- * Drop the tablesync's origin tracking if exists.
- *
- * There is a chance that the user is concurrently performing refresh
- * for the subscription where we remove the table state and its origin
- * or the apply worker would have removed this origin. So passing
- * missing_ok = true.
- */
- replorigin_drop_by_name(originname, true, false);
+ foreach(lc, rstates)
+ {
+ rstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
+ memcpy(rstate, lfirst(lc), sizeof(SubscriptionRelState));
+
+ /*
+ * Pick the table for the next run if there is not another worker
+ * already picked that table.
+ */
+ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+ if (rstate->state != SUBREL_STATE_SYNCDONE &&
+ !logicalrep_worker_find(MySubscription->oid, rstate->relid, false))
+ {
+ /* Update worker state for the next table */
+ MyLogicalRepWorker->is_first_run = false;
+ MyLogicalRepWorker->relid = rstate->relid;
+ MyLogicalRepWorker->relstate = rstate->state;
+ MyLogicalRepWorker->relstate_lsn = rstate->lsn;
+ MyLogicalRepWorker->move_to_next_rel = true;
+ LWLockRelease(LogicalRepWorkerLock);
+ break;
+ }
+ LWLockRelease(LogicalRepWorkerLock);
+ }
+
+ /* Cleanup before next run or ending the worker. */
+ if (!MyLogicalRepWorker->move_to_next_rel)
+ {
+ /*
+ * It is important to give an error if we are unable to drop the
+ * slot, otherwise, it won't be dropped till the corresponding
+ * subscription is dropped. So passing missing_ok = false.
+ */
+ if (MyLogicalRepWorker->created_slot)
+ {
+ /* End streaming if it's not already ended. */
+ if (!is_streaming_ended)
+ walrcv_endstreaming(LogRepWorkerWalRcvConn, &tli);
+ ReplicationSlotDropAtPubNode(LogRepWorkerWalRcvConn, syncslotname, false);
+ }
- finish_sync_worker();
+ /* Drop replication origin before exiting. */
+ replorigin_drop_by_name(originname, true, false);
+
+ finish_sync_worker();
+ }
+ else
+ {
+ clean_sync_worker();
+ }
}
else
SpinLockRelease(&MyLogicalRepWorker->relmutex);
@@ -485,6 +576,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
if (current_lsn >= rstate->lsn)
{
char originname[NAMEDATALEN];
+ bool is_origin_null = true;
rstate->state = SUBREL_STATE_READY;
rstate->lsn = current_lsn;
@@ -505,18 +597,27 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
* error while dropping we won't restart it to drop the
* origin. So passing missing_ok = true.
*/
- ReplicationOriginNameForLogicalRep(MyLogicalRepWorker->subid,
- rstate->relid,
- originname,
- sizeof(originname));
- replorigin_drop_by_name(originname, true, false);
+ GetSubscriptionRelOrigin(MyLogicalRepWorker->subid,
+ rstate->relid, originname,
+ &is_origin_null);
+
+ if (!is_origin_null)
+ {
+ replorigin_drop_by_name(originname, true, false);
+ }
/*
* Update the state to READY only after the origin cleanup.
*/
- UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
- rstate->relid, rstate->state,
- rstate->lsn);
+ UpdateSubscriptionRel(MyLogicalRepWorker->subid,
+ rstate->relid,
+ rstate->state,
+ rstate->lsn,
+ NULL,
+ NULL);
+
+ CommitTransactionCommand();
+ started_tx = false;
}
}
else
@@ -605,11 +706,21 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
TimestampDifferenceExceeds(hentry->last_start_time, now,
wal_retrieve_retry_interval))
{
+ if (IsTransactionState())
+ CommitTransactionCommand();
+ StartTransactionCommand();
+ started_tx = true;
+
+ MySubscription->lastusedid++;
+ UpdateSubscriptionLastSlotId(MyLogicalRepWorker->subid,
+ MySubscription->lastusedid);
+
logicalrep_worker_launch(MyLogicalRepWorker->dbid,
MySubscription->oid,
MySubscription->name,
MyLogicalRepWorker->userid,
- rstate->relid);
+ rstate->relid,
+ MySubscription->lastusedid);
hentry->last_start_time = now;
}
}
@@ -802,6 +913,7 @@ fetch_remote_table_info(char *nspname, char *relname,
TupleTableSlot *tslot;
Oid attrsRow[] = {INT2VECTOROID};
StringInfoData pub_names;
+
initStringInfo(&pub_names);
foreach(lc, MySubscription->publications)
{
@@ -1174,8 +1286,8 @@ copy_table(Relation rel)
* The name must not exceed NAMEDATALEN - 1 because of remote node constraints
* on slot name length. We append system_identifier to avoid slot_name
* collision with subscriptions in other clusters. With the current scheme
- * pg_%u_sync_%u_UINT64_FORMAT (3 + 10 + 6 + 10 + 20 + '\0'), the maximum
- * length of slot_name will be 50.
+ * pg_%u_sync_%lu_UINT64_FORMAT (3 + 10 + 6 + 20 + 20 + '\0'), the maximum
+ * length of slot_name will be 45.
*
* The returned slot name is stored in the supplied buffer (syncslotname) with
* the given size.
@@ -1186,11 +1298,11 @@ copy_table(Relation rel)
* had changed.
*/
void
-ReplicationSlotNameForTablesync(Oid suboid, Oid relid,
+ReplicationSlotNameForTablesync(Oid suboid, int64 slotid,
char *syncslotname, Size szslot)
{
- snprintf(syncslotname, szslot, "pg_%u_sync_%u_" UINT64_FORMAT, suboid,
- relid, GetSystemIdentifier());
+ snprintf(syncslotname, szslot, "pg_%u_sync_%ld_" UINT64_FORMAT, suboid,
+ slotid, GetSystemIdentifier());
}
/*
@@ -1213,6 +1325,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
WalRcvExecResult *res;
char originname[NAMEDATALEN];
RepOriginId originid;
+ char *prev_slotname;
/* Check the state of the table synchronization. */
StartTransactionCommand();
@@ -1241,7 +1354,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
/* Calculate the name of the tablesync slot. */
slotname = (char *) palloc(NAMEDATALEN);
ReplicationSlotNameForTablesync(MySubscription->oid,
- MyLogicalRepWorker->relid,
+ MyLogicalRepWorker->rep_slot_id,
slotname,
NAMEDATALEN);
@@ -1261,11 +1374,25 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
MyLogicalRepWorker->relstate == SUBREL_STATE_DATASYNC ||
MyLogicalRepWorker->relstate == SUBREL_STATE_FINISHEDCOPY);
+ /*
+ * See if tablesync of the current relation has been started with another
+ * replication slot.
+ *
+ * Read previous slot name from the catalog, if exists.
+ */
+ prev_slotname = (char *) palloc0(NAMEDATALEN);
+ StartTransactionCommand();
+ GetSubscriptionRelReplicationSlot(MyLogicalRepWorker->subid,
+ MyLogicalRepWorker->relid,
+ prev_slotname);
+
/* Assign the origin tracking record name. */
ReplicationOriginNameForLogicalRep(MySubscription->oid,
- MyLogicalRepWorker->relid,
originname,
- sizeof(originname));
+ sizeof(originname),
+ true);
+
+ CommitTransactionCommand();
if (MyLogicalRepWorker->relstate == SUBREL_STATE_DATASYNC)
{
@@ -1280,10 +1407,48 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
* breakdown then it wouldn't have succeeded so trying it next time
* seems like a better bet.
*/
- ReplicationSlotDropAtPubNode(LogRepWorkerWalRcvConn, slotname, true);
+ if (strlen(prev_slotname) > 0)
+ {
+ ReplicationSlotDropAtPubNode(LogRepWorkerWalRcvConn, prev_slotname, true);
+
+ StartTransactionCommand();
+ /* Replication drop might still exist. Try to drop */
+ replorigin_drop_by_name(originname, true, false);
+
+ /*
+ * Remove replication slot and origin name from the relation's
+ * catalog record
+ */
+ UpdateSubscriptionRel(MyLogicalRepWorker->subid,
+ MyLogicalRepWorker->relid,
+ MyLogicalRepWorker->relstate,
+ MyLogicalRepWorker->relstate_lsn,
+ NULL,
+ NULL);
+ CommitTransactionCommand();
+ }
}
else if (MyLogicalRepWorker->relstate == SUBREL_STATE_FINISHEDCOPY)
{
+ /*
+ * At this point, the table that is currently being synchronized
+ * should have its replication slot name filled in the catalog. The
+ * tablesync process was started with another sync worker and
+ * replication slot. We need to continue using the same replication
+ * slot in this worker too.
+ */
+ if (strlen(prev_slotname) == 0)
+ {
+ elog(ERROR, "Replication slot could not be found for relation %u",
+ MyLogicalRepWorker->relid);
+ }
+
+ /*
+ * Proceed with the correct replication slot. Use previously created
+ * replication slot to sync this table.
+ */
+ slotname = prev_slotname;
+
/*
* The COPY phase was previously done, but tablesync then crashed
* before it was able to finish normally.
@@ -1304,6 +1469,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
goto copy_table_done;
}
+ /* Preparing for table copy operation */
SpinLockAcquire(&MyLogicalRepWorker->relmutex);
MyLogicalRepWorker->relstate = SUBREL_STATE_DATASYNC;
MyLogicalRepWorker->relstate_lsn = InvalidXLogRecPtr;
@@ -1311,10 +1477,12 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
/* Update the state and make it visible to others. */
StartTransactionCommand();
- UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
- MyLogicalRepWorker->relid,
- MyLogicalRepWorker->relstate,
- MyLogicalRepWorker->relstate_lsn);
+ UpdateSubscriptionRel(MyLogicalRepWorker->subid,
+ MyLogicalRepWorker->relid,
+ MyLogicalRepWorker->relstate,
+ MyLogicalRepWorker->relstate_lsn,
+ slotname,
+ originname);
CommitTransactionCommand();
pgstat_report_stat(true);
@@ -1353,6 +1521,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
GetUserNameFromId(GetUserId(), true),
RelationGetRelationName(rel))));
+
/*
* Start a transaction in the remote node in REPEATABLE READ mode. This
* ensures that both the replication slot we create (see below) and the
@@ -1368,55 +1537,100 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
res->err)));
walrcv_clear_result(res);
+ originid = replorigin_by_name(originname, true);
+
/*
* Create a new permanent logical decoding slot. This slot will be used
* for the catchup phase after COPY is done, so tell it to use the
* snapshot to make the final data consistent.
*
+ * Replication slot will only be created if either this is the first run
+ * of the worker or we're not using a previous replication slot.
+ *
* Prevent cancel/die interrupts while creating slot here because it is
* possible that before the server finishes this command, a concurrent
* drop subscription happens which would complete without removing this
* slot leading to a dangling slot on the server.
+ *
*/
- HOLD_INTERRUPTS();
- walrcv_create_slot(LogRepWorkerWalRcvConn,
- slotname, false /* permanent */ , false /* two_phase */ ,
- CRS_USE_SNAPSHOT, origin_startpos);
- RESUME_INTERRUPTS();
-
- /*
- * Setup replication origin tracking. The purpose of doing this before the
- * copy is to avoid doing the copy again due to any error in setting up
- * origin tracking.
- */
- originid = replorigin_by_name(originname, true);
- if (!OidIsValid(originid))
+ if (!MyLogicalRepWorker->created_slot)
{
+ HOLD_INTERRUPTS();
+ walrcv_create_slot(LogRepWorkerWalRcvConn,
+ slotname, false /* permanent */ , false /* two_phase */ ,
+ CRS_USE_SNAPSHOT, origin_startpos);
+ RESUME_INTERRUPTS();
+
/*
- * Origin tracking does not exist, so create it now.
- *
- * Then advance to the LSN got from walrcv_create_slot. This is WAL
- * logged for the purpose of recovery. Locks are to prevent the
- * replication origin from vanishing while advancing.
+ * Remember that we created the slot so that we will not try to create
+ * it again.
*/
- originid = replorigin_create(originname);
+ SpinLockAcquire(&MyLogicalRepWorker->relmutex);
+ MyLogicalRepWorker->created_slot = true;
+ SpinLockRelease(&MyLogicalRepWorker->relmutex);
- LockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
- replorigin_advance(originid, *origin_startpos, InvalidXLogRecPtr,
- true /* go backward */ , true /* WAL log */ );
- UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
+ /*
+ * Setup replication origin tracking. The purpose of doing this before
+ * the copy is to avoid doing the copy again due to any error in
+ * setting up origin tracking.
+ */
- replorigin_session_setup(originid);
- replorigin_session_origin = originid;
+ if (!OidIsValid(originid))
+ {
+ /*
+ * Origin tracking does not exist, so create it now.
+ */
+ originid = replorigin_create(originname);
+ }
+ else
+ {
+ /*
+ * At this point, there shouldn't be any existing replication
+ * origin wit the same name.
+ */
+ ereport(ERROR,
+ (errcode(ERRCODE_DUPLICATE_OBJECT),
+ errmsg("replication origin \"%s\" already exists",
+ originname)));
+ }
}
else
{
- ereport(ERROR,
- (errcode(ERRCODE_DUPLICATE_OBJECT),
- errmsg("replication origin \"%s\" already exists",
- originname)));
+ /*
+ * Do not create a new replication slot, reuse the existing one
+ * instead. Use a new snapshot for the replication slot to ensure that
+ * tablesync and apply proceses are consistent with each other.
+ */
+ WalRcvStreamOptions options;
+ int server_version;
+
+ server_version = walrcv_server_version(LogRepWorkerWalRcvConn);
+ options.proto.logical.proto_version =
+ server_version >= 150000 ? LOGICALREP_PROTO_TWOPHASE_VERSION_NUM :
+ server_version >= 140000 ? LOGICALREP_PROTO_STREAM_VERSION_NUM :
+ LOGICALREP_PROTO_VERSION_NUM;
+ options.proto.logical.publication_names = MySubscription->publications;
+
+ HOLD_INTERRUPTS();
+ walrcv_slot_snapshot(LogRepWorkerWalRcvConn, slotname, &options, origin_startpos);
+ RESUME_INTERRUPTS();
}
+ /*
+ * Advance to the LSN got from walrcv_create_slot. This is WAL
+ * logged for the purpose of recovery. Locks are to prevent the
+ * replication origin from vanishing while advancing.
+ *
+ * Then setup replication origin tracking.
+ */
+ LockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
+ replorigin_advance(originid, *origin_startpos, InvalidXLogRecPtr,
+ true /* go backward */ , true /* WAL log */ );
+ UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
+
+ replorigin_session_setup(originid);
+ replorigin_session_origin = originid;
+
/* Now do the initial data copy */
PushActiveSnapshot(GetTransactionSnapshot());
copy_table(rel);
@@ -1439,10 +1653,12 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
* Update the persisted state to indicate the COPY phase is done; make it
* visible to others.
*/
- UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
- MyLogicalRepWorker->relid,
- SUBREL_STATE_FINISHEDCOPY,
- MyLogicalRepWorker->relstate_lsn);
+ UpdateSubscriptionRel(MyLogicalRepWorker->subid,
+ MyLogicalRepWorker->relid,
+ SUBREL_STATE_FINISHEDCOPY,
+ MyLogicalRepWorker->relstate_lsn,
+ slotname,
+ originname);
CommitTransactionCommand();
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 96772e4d73..0b325b8fa0 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -317,6 +317,7 @@ static void stream_open_file(Oid subid, TransactionId xid,
bool first_segment);
static void stream_write_change(char action, StringInfo s);
static void stream_close_file(void);
+static void stream_build_options(WalRcvStreamOptions *options, char *slotname, XLogRecPtr *origin_startpos);
static void send_feedback(XLogRecPtr recvpos, bool force, bool requestReply);
@@ -369,18 +370,26 @@ static inline void reset_apply_error_context_info(void);
* Form the origin name for the subscription.
*
* This is a common function for tablesync and other workers. Tablesync workers
- * must pass a valid relid. Other callers must pass relid = InvalidOid.
+ * must pass is_tablesync true so that origin name includes slot id.
*
* Return the name in the supplied buffer.
*/
void
-ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,
- char *originname, Size szoriginname)
+ReplicationOriginNameForLogicalRep(Oid suboid, char *originname,
+ Size szoriginname, bool is_tablesync)
{
- if (OidIsValid(relid))
+ if (is_tablesync)
{
- /* Replication origin name for tablesync workers. */
- snprintf(originname, szoriginname, "pg_%u_%u", suboid, relid);
+ bool is_null = true;
+
+ /*
+ * Replication origin name for tablesync workers. First, look into the
+ * catalog. If originname does not exist, then use the default name.
+ */
+ GetSubscriptionRelOrigin(suboid, MyLogicalRepWorker->relid,
+ originname, &is_null);
+ if (is_null)
+ snprintf(originname, szoriginname, "pg_%u_%ld", suboid, MyLogicalRepWorker->rep_slot_id);
}
else
{
@@ -2839,6 +2848,23 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
MemoryContextReset(ApplyMessageContext);
}
+ /*
+ * apply_dispatch() may have gone into apply_handle_commit()
+ * which can move to next table while running
+ * process_syncing_tables_for_sync. Before we were able to
+ * reuse tablesync workers, that
+ * process_syncing_tables_for_sync call would exit the worker
+ * instead of moving to next table. Now that tablesync workers
+ * can be reused, we need to take care of memory contexts here
+ * before moving to sync a table.
+ */
+ if (MyLogicalRepWorker->move_to_next_rel)
+ {
+ MemoryContextResetAndDeleteChildren(ApplyMessageContext);
+ MemoryContextSwitchTo(TopMemoryContext);
+ return;
+ }
+
len = walrcv_receive(LogRepWorkerWalRcvConn, &buf, &fd);
}
}
@@ -2858,6 +2884,10 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
/* Process any table synchronization changes. */
process_syncing_tables(last_received);
+ if (MyLogicalRepWorker->move_to_next_rel)
+ {
+ endofstream = true;
+ }
}
/* Cleanup the memory. */
@@ -2959,8 +2989,16 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
/* Pop the error context stack */
error_context_stack = errcallback.previous;
- /* All done */
- walrcv_endstreaming(LogRepWorkerWalRcvConn, &tli);
+ /*
+ * If it's moving to next relation, this is a sync worker. Sync workers
+ * end the streaming during process_syncing_tables_for_sync. Calling
+ * endstreaming twice causes "no COPY in progress" errors.
+ */
+ if (!MyLogicalRepWorker->move_to_next_rel)
+ {
+ /* All done */
+ walrcv_endstreaming(LogRepWorkerWalRcvConn, &tli);
+ }
}
/*
@@ -3502,6 +3540,35 @@ stream_write_change(char action, StringInfo s)
BufFileWrite(stream_fd, &s->data[s->cursor], len);
}
+/*
+ * stream_build_options_replication
+ * Build logical replication streaming options.
+ *
+ * This function sets streaming options including replication slot name
+ * and origin start position. Workers need these options for logical replication.
+ */
+static void
+stream_build_options(WalRcvStreamOptions *options, char *slotname, XLogRecPtr *origin_startpos)
+{
+ int server_version;
+
+ options->logical = true;
+ options->startpoint = *origin_startpos;
+ options->slotname = slotname;
+
+ server_version = walrcv_server_version(LogRepWorkerWalRcvConn);
+ options->proto.logical.proto_version =
+ server_version >= 150000 ? LOGICALREP_PROTO_TWOPHASE_VERSION_NUM :
+ server_version >= 140000 ? LOGICALREP_PROTO_STREAM_VERSION_NUM :
+ LOGICALREP_PROTO_VERSION_NUM;
+
+ options->proto.logical.publication_names = MySubscription->publications;
+ options->proto.logical.binary = MySubscription->binary;
+ options->proto.logical.streaming = MySubscription->stream;
+ options->proto.logical.twophase = false;
+ options->proto.logical.origin = pstrdup(MySubscription->origin);
+}
+
/*
* Cleanup the memory for subxacts and reset the related variables.
*/
@@ -3576,6 +3643,9 @@ start_table_sync(XLogRecPtr *origin_startpos, char **myslotname)
/* allocate slot name in long-lived context */
*myslotname = MemoryContextStrdup(ApplyContext, syncslotname);
+
+ /* Keep the replication slot name used for this sync. */
+ MyLogicalRepWorker->slot_name = *myslotname;
pfree(syncslotname);
}
@@ -3613,6 +3683,140 @@ start_apply(XLogRecPtr origin_startpos)
PG_END_TRY();
}
+/*
+ * Runs the tablesync worker.
+ * It starts table sync. After successful sync,
+ * builds streaming options and starts streaming.
+ */
+static void
+run_tablesync_worker(WalRcvStreamOptions *options,
+ char *slotname,
+ char *originname,
+ int originname_size,
+ XLogRecPtr *origin_startpos)
+{
+ /* Set this to false for safety, in case we're already reusing the worker */
+ MyLogicalRepWorker->move_to_next_rel = false;
+
+ start_table_sync(origin_startpos, &slotname);
+
+ /*
+ * Allocate the origin name in long-lived context for error context
+ * message.
+ */
+ StartTransactionCommand();
+ ReplicationOriginNameForLogicalRep(MySubscription->oid,
+ originname,
+ originname_size,
+ true);
+ CommitTransactionCommand();
+
+ apply_error_callback_arg.origin_name = MemoryContextStrdup(ApplyContext,
+ originname);
+
+ stream_build_options(options, slotname, origin_startpos);
+
+ /* Start normal logical streaming replication. */
+ walrcv_startstreaming(LogRepWorkerWalRcvConn, options);
+}
+
+/*
+ * Runs the apply worker.
+ * It sets up replication origin, the streaming options
+ * and then starts streaming.
+ */
+static void
+run_apply_worker(WalRcvStreamOptions *options,
+ char *slotname,
+ char *originname,
+ int originname_size,
+ XLogRecPtr *origin_startpos)
+{
+ RepOriginId originid;
+ TimeLineID startpointTLI;
+ char *err;
+
+ slotname = MySubscription->slotname;
+
+ /*
+ * This shouldn't happen if the subscription is enabled, but guard against
+ * DDL bugs or manual catalog changes. (libpqwalreceiver will crash if
+ * slot is NULL.)
+ */
+ if (!slotname)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("subscription has no replication slot set")));
+
+ /* Setup replication origin tracking. */
+ StartTransactionCommand();
+ ReplicationOriginNameForLogicalRep(MySubscription->oid, originname,
+ originname_size, false);
+ originid = replorigin_by_name(originname, true);
+ if (!OidIsValid(originid))
+ originid = replorigin_create(originname);
+ replorigin_session_setup(originid);
+ replorigin_session_origin = originid;
+ *origin_startpos = replorigin_session_get_progress(false);
+ CommitTransactionCommand();
+
+ LogRepWorkerWalRcvConn = walrcv_connect(MySubscription->conninfo, true,
+ MySubscription->name, &err);
+ if (LogRepWorkerWalRcvConn == NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_CONNECTION_FAILURE),
+ errmsg("could not connect to the publisher: %s", err)));
+
+ /*
+ * We don't really use the output identify_system for anything but it does
+ * some initializations on the upstream so let's still call it.
+ */
+ (void) walrcv_identify_system(LogRepWorkerWalRcvConn, &startpointTLI);
+
+ /*
+ * Allocate the origin name in long-lived context for error context
+ * message.
+ */
+ apply_error_callback_arg.origin_name = MemoryContextStrdup(ApplyContext,
+ originname);
+
+ stream_build_options(options, slotname, origin_startpos);
+
+ /*
+ * Even when the two_phase mode is requested by the user, it remains as
+ * the tri-state PENDING until all tablesyncs have reached READY state.
+ * Only then, can it become ENABLED.
+ *
+ * Note: If the subscription has no tables then leave the state as
+ * PENDING, which allows ALTER SUBSCRIPTION ... REFRESH PUBLICATION to
+ * work.
+ */
+ if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING &&
+ AllTablesyncsReady())
+ {
+ /* Start streaming with two_phase enabled */
+ options->proto.logical.twophase = true;
+ walrcv_startstreaming(LogRepWorkerWalRcvConn, options);
+
+ StartTransactionCommand();
+ UpdateTwoPhaseState(MySubscription->oid, LOGICALREP_TWOPHASE_STATE_ENABLED);
+ MySubscription->twophasestate = LOGICALREP_TWOPHASE_STATE_ENABLED;
+ CommitTransactionCommand();
+ }
+ else
+ {
+ walrcv_startstreaming(LogRepWorkerWalRcvConn, options);
+ }
+
+ ereport(DEBUG1,
+ (errmsg_internal("logical replication apply worker for subscription \"%s\" two_phase is %s",
+ MySubscription->name,
+ MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_DISABLED ? "DISABLED" :
+ MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING ? "PENDING" :
+ MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED ? "ENABLED" :
+ "?")));
+}
+
/* Logical Replication Apply worker entry point */
void
ApplyWorkerMain(Datum main_arg)
@@ -3623,7 +3827,6 @@ ApplyWorkerMain(Datum main_arg)
XLogRecPtr origin_startpos = InvalidXLogRecPtr;
char *myslotname = NULL;
WalRcvStreamOptions options;
- int server_version;
/* Attach to slot */
logicalrep_worker_attach(worker_slot);
@@ -3714,143 +3917,48 @@ ApplyWorkerMain(Datum main_arg)
elog(DEBUG1, "connecting to publisher using connection string \"%s\"",
MySubscription->conninfo);
- if (am_tablesync_worker())
- {
- start_table_sync(&origin_startpos, &myslotname);
-
- /*
- * Allocate the origin name in long-lived context for error context
- * message.
- */
- ReplicationOriginNameForLogicalRep(MySubscription->oid,
- MyLogicalRepWorker->relid,
- originname,
- sizeof(originname));
- apply_error_callback_arg.origin_name = MemoryContextStrdup(ApplyContext,
- originname);
- }
- else
- {
- /* This is main apply worker */
- RepOriginId originid;
- TimeLineID startpointTLI;
- char *err;
-
- myslotname = MySubscription->slotname;
-
- /*
- * This shouldn't happen if the subscription is enabled, but guard
- * against DDL bugs or manual catalog changes. (libpqwalreceiver will
- * crash if slot is NULL.)
- */
- if (!myslotname)
- ereport(ERROR,
- (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
- errmsg("subscription has no replication slot set")));
-
- /* Setup replication origin tracking. */
- StartTransactionCommand();
- ReplicationOriginNameForLogicalRep(MySubscription->oid, InvalidOid,
- originname, sizeof(originname));
- originid = replorigin_by_name(originname, true);
- if (!OidIsValid(originid))
- originid = replorigin_create(originname);
- replorigin_session_setup(originid);
- replorigin_session_origin = originid;
- origin_startpos = replorigin_session_get_progress(false);
- CommitTransactionCommand();
-
- LogRepWorkerWalRcvConn = walrcv_connect(MySubscription->conninfo, true,
- MySubscription->name, &err);
- if (LogRepWorkerWalRcvConn == NULL)
- ereport(ERROR,
- (errcode(ERRCODE_CONNECTION_FAILURE),
- errmsg("could not connect to the publisher: %s", err)));
-
- /*
- * We don't really use the output identify_system for anything but it
- * does some initializations on the upstream so let's still call it.
- */
- (void) walrcv_identify_system(LogRepWorkerWalRcvConn, &startpointTLI);
-
- /*
- * Allocate the origin name in long-lived context for error context
- * message.
- */
- apply_error_callback_arg.origin_name = MemoryContextStrdup(ApplyContext,
- originname);
- }
-
/*
* Setup callback for syscache so that we know when something changes in
- * the subscription relation state.
+ * the subscription relation state. Do this outside the loop to avoid
+ * exceeding MAX_SYSCACHE_CALLBACKS
*/
CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
invalidate_syncing_table_states,
(Datum) 0);
- /* Build logical replication streaming options. */
- options.logical = true;
- options.startpoint = origin_startpos;
- options.slotname = myslotname;
-
- server_version = walrcv_server_version(LogRepWorkerWalRcvConn);
- options.proto.logical.proto_version =
- server_version >= 150000 ? LOGICALREP_PROTO_TWOPHASE_VERSION_NUM :
- server_version >= 140000 ? LOGICALREP_PROTO_STREAM_VERSION_NUM :
- LOGICALREP_PROTO_VERSION_NUM;
-
- options.proto.logical.publication_names = MySubscription->publications;
- options.proto.logical.binary = MySubscription->binary;
- options.proto.logical.streaming = MySubscription->stream;
- options.proto.logical.twophase = false;
- options.proto.logical.origin = pstrdup(MySubscription->origin);
-
- if (!am_tablesync_worker())
+ /*
+ * The loop where worker does its job. It loops until the worker is not
+ * reused.
+ */
+ while (MyLogicalRepWorker->is_first_run ||
+ MyLogicalRepWorker->move_to_next_rel)
{
- /*
- * Even when the two_phase mode is requested by the user, it remains
- * as the tri-state PENDING until all tablesyncs have reached READY
- * state. Only then, can it become ENABLED.
- *
- * Note: If the subscription has no tables then leave the state as
- * PENDING, which allows ALTER SUBSCRIPTION ... REFRESH PUBLICATION to
- * work.
- */
- if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING &&
- AllTablesyncsReady())
+ if (am_tablesync_worker())
{
- /* Start streaming with two_phase enabled */
- options.proto.logical.twophase = true;
- walrcv_startstreaming(LogRepWorkerWalRcvConn, &options);
-
- StartTransactionCommand();
- UpdateTwoPhaseState(MySubscription->oid, LOGICALREP_TWOPHASE_STATE_ENABLED);
- MySubscription->twophasestate = LOGICALREP_TWOPHASE_STATE_ENABLED;
- CommitTransactionCommand();
+ /*
+ * This is a tablesync worker. Start syncing tables before
+ * starting the apply loop.
+ */
+ run_tablesync_worker(&options, myslotname, originname, sizeof(originname), &origin_startpos);
}
else
{
- walrcv_startstreaming(LogRepWorkerWalRcvConn, &options);
+ /* This is main apply worker */
+ run_apply_worker(&options, myslotname, originname, sizeof(originname), &origin_startpos);
}
- ereport(DEBUG1,
- (errmsg_internal("logical replication apply worker for subscription \"%s\" two_phase is %s",
- MySubscription->name,
- MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_DISABLED ? "DISABLED" :
- MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING ? "PENDING" :
- MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED ? "ENABLED" :
- "?")));
- }
- else
- {
- /* Start normal logical streaming replication. */
- walrcv_startstreaming(LogRepWorkerWalRcvConn, &options);
- }
-
- /* Run the main loop. */
- start_apply(origin_startpos);
+ /* Run the main loop. */
+ start_apply(origin_startpos);
+ if (MyLogicalRepWorker->move_to_next_rel)
+ {
+ StartTransactionCommand();
+ ereport(LOG,
+ (errmsg("logical replication table synchronization worker for subscription \"%s\" has moved to sync table \"%s\".",
+ MySubscription->name, get_rel_name(MyLogicalRepWorker->relid))));
+ CommitTransactionCommand();
+ }
+ }
proc_exit(0);
}
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index 7b98714f30..0b13dd34cb 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -102,6 +102,9 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
/* Only publish data originating from the specified origin */
text suborigin BKI_DEFAULT(LOGICALREP_ORIGIN_ANY);
+
+ /* The last used ID to create a replication slot for tablesync */
+ int64 sublastusedid BKI_DEFAULT(0);
#endif
} FormData_pg_subscription;
@@ -135,11 +138,14 @@ typedef struct Subscription
List *publications; /* List of publication names to subscribe to */
char *origin; /* Only publish data originating from the
* specified origin */
+ int64 lastusedid; /* Last used unique ID to create replication
+ * slots in tablesync */
} Subscription;
extern Subscription *GetSubscription(Oid subid, bool missing_ok);
extern void FreeSubscription(Subscription *sub);
extern void DisableSubscription(Oid subid);
+extern void UpdateSubscriptionLastSlotId(Oid subid, int64 lastusedid);
extern int CountDBSubscriptions(Oid dbid);
diff --git a/src/include/catalog/pg_subscription_rel.h b/src/include/catalog/pg_subscription_rel.h
index 8e88de7b2b..9cafb521f9 100644
--- a/src/include/catalog/pg_subscription_rel.h
+++ b/src/include/catalog/pg_subscription_rel.h
@@ -44,6 +44,12 @@ CATALOG(pg_subscription_rel,6102,SubscriptionRelRelationId)
* used for synchronization
* coordination, or NULL if not
* valid */
+ text srrelslotname BKI_FORCE_NULL; /* name of the replication
+ * slot for relation in
+ * subscription */
+ text srreloriginname BKI_FORCE_NULL; /* origin name for relation in
+ * subscription */
+
#endif
} FormData_pg_subscription_rel;
@@ -81,10 +87,17 @@ typedef struct SubscriptionRelState
} SubscriptionRelState;
extern void AddSubscriptionRelState(Oid subid, Oid relid, char state,
- XLogRecPtr sublsn);
+ XLogRecPtr sublsn, char *relslotname, char *reloriginname);
extern void UpdateSubscriptionRelState(Oid subid, Oid relid, char state,
XLogRecPtr sublsn);
+extern void UpdateSubscriptionRel(Oid subid, Oid relid, char state,
+ XLogRecPtr sublsn, char *relslotname, char *reloriginname);
+extern void UpdateSubscriptionRelReplicationSlot(Oid subid, Oid relid, char *relslotname);
+
extern char GetSubscriptionRelState(Oid subid, Oid relid, XLogRecPtr *sublsn);
+extern void GetSubscriptionRelReplicationSlot(Oid subid, Oid relid, char *slotname);
+extern void GetSubscriptionRelOrigin(Oid subid, Oid relid, char *reloriginname, bool *isnull);
+
extern void RemoveSubscriptionRel(Oid subid, Oid relid);
extern bool HasSubscriptionRelations(Oid subid);
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 65f2c74239..aca6b590f9 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -219,8 +219,9 @@ extern bool InvalidateObsoleteReplicationSlots(XLogSegNo oldestSegno);
extern ReplicationSlot *SearchNamedReplicationSlot(const char *name, bool need_lock);
extern int ReplicationSlotIndex(ReplicationSlot *slot);
extern bool ReplicationSlotName(int index, Name name);
-extern void ReplicationSlotNameForTablesync(Oid suboid, Oid relid, char *syncslotname, Size szslot);
+extern void ReplicationSlotNameForTablesync(Oid suboid, int64 slotid, char *syncslotname, Size szslot);
extern void ReplicationSlotDropAtPubNode(WalReceiverConn *wrconn, char *slotname, bool missing_ok);
+extern List *GetReplicationSlotNamesBySubId(WalReceiverConn *wrconn, Oid subid, bool missing_ok);
extern void StartupReplicationSlots(void);
extern void CheckPointReplicationSlots(void);
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 2b7114ff6d..e7f74e0f9f 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -30,6 +30,26 @@ typedef struct LogicalRepWorker
/* Indicates if this slot is used or free. */
bool in_use;
+ /*
+ * Indicates if worker is running for the first time or in reuse
+ */
+ bool is_first_run;
+
+ /*
+ * Indicates if the sync worker created a replication slot or it reuses an
+ * existing one created by another worker.
+ */
+ bool created_slot;
+
+ /*
+ * Unique identifier for replication slot to be created by tablesnync
+ * workers, if needed.
+ */
+ int64 rep_slot_id;
+
+ /* Replication slot name used by the worker. */
+ char *slot_name;
+
/* Increased every time the slot is taken by new worker. */
uint16 generation;
@@ -51,6 +71,12 @@ typedef struct LogicalRepWorker
XLogRecPtr relstate_lsn;
slock_t relmutex;
+ /*
+ * Used to indicate whether sync worker will be reused for another
+ * relation
+ */
+ bool move_to_next_rel;
+
/*
* Used to create the changes and subxact files for the streaming
* transactions. Upon the arrival of the first streaming transaction, the
@@ -85,15 +111,15 @@ extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
bool only_running);
extern List *logicalrep_workers_find(Oid subid, bool only_running);
extern void logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,
- Oid userid, Oid relid);
+ Oid userid, Oid relid, int64 slotid);
extern void logicalrep_worker_stop(Oid subid, Oid relid);
extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
extern int logicalrep_sync_worker_count(Oid subid);
-extern void ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,
- char *originname, Size szoriginname);
+extern void ReplicationOriginNameForLogicalRep(Oid suboid, char *originname,
+ Size szoriginname, bool is_tablesync);
extern char *LogicalRepSyncTableStart(XLogRecPtr *origin_startpos);
extern bool AllTablesyncsReady(void);
diff --git a/src/test/regress/expected/misc_sanity.out b/src/test/regress/expected/misc_sanity.out
index a57fd142a9..3d34a21421 100644
--- a/src/test/regress/expected/misc_sanity.out
+++ b/src/test/regress/expected/misc_sanity.out
@@ -47,20 +47,22 @@ WHERE c.oid < 16384 AND
relkind = 'r' AND
attstorage != 'p'
ORDER BY 1, 2;
- relname | attname | atttypid
--------------------------+---------------+--------------
- pg_attribute | attacl | aclitem[]
- pg_attribute | attfdwoptions | text[]
- pg_attribute | attmissingval | anyarray
- pg_attribute | attoptions | text[]
- pg_class | relacl | aclitem[]
- pg_class | reloptions | text[]
- pg_class | relpartbound | pg_node_tree
- pg_index | indexprs | pg_node_tree
- pg_index | indpred | pg_node_tree
- pg_largeobject | data | bytea
- pg_largeobject_metadata | lomacl | aclitem[]
-(11 rows)
+ relname | attname | atttypid
+-------------------------+-----------------+--------------
+ pg_attribute | attacl | aclitem[]
+ pg_attribute | attfdwoptions | text[]
+ pg_attribute | attmissingval | anyarray
+ pg_attribute | attoptions | text[]
+ pg_class | relacl | aclitem[]
+ pg_class | reloptions | text[]
+ pg_class | relpartbound | pg_node_tree
+ pg_index | indexprs | pg_node_tree
+ pg_index | indpred | pg_node_tree
+ pg_largeobject | data | bytea
+ pg_largeobject_metadata | lomacl | aclitem[]
+ pg_subscription_rel | srreloriginname | text
+ pg_subscription_rel | srrelslotname | text
+(13 rows)
-- system catalogs without primary keys
--
--
2.25.1
view thread (24+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected]
Subject: Re: [PATCH] Reuse Workers and Replication Slots during Logical Replication
In-Reply-To: <CAGPVpCQdZ_oj-QFcTOhTrUTs-NCKrrZ=ZNCNPR1qe27rXV-iYw@mail.gmail.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox