Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commit208c5d6

Browse files
author
Amit Kapila
committed
Add ALTER SUBSCRIPTION ... SKIP.
This feature allows skipping the transaction on subscriber nodes.If incoming change violates any constraint, logical replication stopsuntil it's resolved. Currently, users need to either manually resolve theconflict by updating a subscriber-side database or by using functionpg_replication_origin_advance() to skip the conflicting transaction. Thiscommit introduces a simpler way to skip the conflicting transactions.The user can specify LSN by ALTER SUBSCRIPTION ... SKIP (lsn = XXX),which allows the apply worker to skip the transaction finished atspecified LSN. The apply worker skips all data modification changes withinthe transaction.Author: Masahiko SawadaReviewed-by: Takamichi Osumi, Hou Zhijie, Peter Eisentraut, Amit Kapila, Shi Yu, Vignesh C, Greg Nancarrow, Haiying Tang, Euler TaveiraDiscussion:https://postgr.es/m/CAD21AoDeScrsHhLyEPYqN3sydg6PxAPVBboK=30xJfUVihNZDA@mail.gmail.com
1 parent315ae75 commit208c5d6

File tree

18 files changed

+665
-173
lines changed

18 files changed

+665
-173
lines changed

‎doc/src/sgml/catalogs.sgml

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7797,6 +7797,16 @@ SCRAM-SHA-256$<replaceable>&lt;iteration count&gt;</replaceable>:<replaceable>&l
77977797
</para></entry>
77987798
</row>
77997799

7800+
<row>
7801+
<entry role="catalog_table_entry"><para role="column_definition">
7802+
<structfield>subskiplsn</structfield> <type>pg_lsn</type>
7803+
</para>
7804+
<para>
7805+
Finish LSN of the transaction whose changes are to be skipped, if a valid
7806+
LSN; otherwise <literal>0/0</literal>.
7807+
</para></entry>
7808+
</row>
7809+
78007810
<row>
78017811
<entry role="catalog_table_entry"><para role="column_definition">
78027812
<structfield>subconninfo</structfield> <type>text</type>

‎doc/src/sgml/logical-replication.sgml

Lines changed: 16 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -362,19 +362,24 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER
362362
</screen>
363363
The LSN of the transaction that contains the change violating the constraint and
364364
the replication origin name can be found from the server log (LSN 0/14C0378 and
365-
replication origin <literal>pg_16395</literal> in the above case). To skip the
366-
transaction, the subscription needs to be disabled temporarily by
367-
<command>ALTER SUBSCRIPTION ... DISABLE</command> first or alternatively, the
365+
replication origin <literal>pg_16395</literal> in the above case). The
366+
transaction that produces conflict can be skipped by using
367+
<command>ALTER SUBSCRIPTION ... SKIP</command> with the finish LSN
368+
(i.e., LSN 0/14C0378). The finish LSN could be an LSN at which the transaction
369+
is committed or prepared on the publisher. Alternatively, the transaction can
370+
also be skipped by calling the <link linkend="pg-replication-origin-advance">
371+
<function>pg_replication_origin_advance()</function></link> function
372+
transaction. Before using this function, the subscription needs to be disabled
373+
temporarily either by <command>ALTER SUBSCRIPTION ... DISABLE</command> or, the
368374
subscription can be used with the <literal>disable_on_error</literal> option.
369-
Then, the transaction can be skipped by calling the
370-
<link linkend="pg-replication-origin-advance">
371-
<function>pg_replication_origin_advance()</function></link> function with
372-
the <parameter>node_name</parameter> (i.e., <literal>pg_16395</literal>) and the
373-
next LSN of the transaction's LSN (i.e., LSN 0/14C0379). After that the replication
374-
can be resumed by <command>ALTER SUBSCRIPTION ... ENABLE</command>. The current
375-
position of origins can be seen in the
376-
<link linkend="view-pg-replication-origin-status">
375+
Then, you can use <function>pg_replication_origin_advance()</function> function
376+
with the <parameter>node_name</parameter> (i.e., <literal>pg_16395</literal>)
377+
and the next LSN of the finish LSN (i.e., 0/14C0379). The current position of
378+
origins can be seen in the <link linkend="view-pg-replication-origin-status">
377379
<structname>pg_replication_origin_status</structname></link> system view.
380+
Please note that skipping the whole transaction include skipping changes that
381+
might not violate any constraint. This can easily make the subscriber
382+
inconsistent.
378383
</para>
379384
</sect1>
380385

‎doc/src/sgml/ref/alter_subscription.sgml

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> REFRESH PUB
2929
ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> ENABLE
3030
ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> DISABLE
3131
ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> SET ( <replaceable class="parameter">subscription_parameter</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] )
32+
ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> SKIP ( <replaceable class="parameter">skip_option</replaceable> = <replaceable class="parameter">value</replaceable> )
3233
ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> OWNER TO { <replaceable>new_owner</replaceable> | CURRENT_ROLE | CURRENT_USER | SESSION_USER }
3334
ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <replaceable>new_name</replaceable>
3435
</synopsis>
@@ -210,6 +211,47 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
210211
</listitem>
211212
</varlistentry>
212213

214+
<varlistentry>
215+
<term><literal>SKIP ( <replaceable class="parameter">skip_option</replaceable> = <replaceable class="parameter">value</replaceable> )</literal></term>
216+
<listitem>
217+
<para>
218+
Skips applying all changes of the remote transaction. If incoming data
219+
violates any constraints, logical replication will stop until it is
220+
resolved. By using <command>ALTER SUBSCRIPTION ... SKIP</command> command,
221+
the logical replication worker skips all data modification changes within
222+
the transaction. This option has no effect on the transactions that are
223+
already prepared by enabling <literal>two_phase</literal> on
224+
subscriber.
225+
After logical replication worker successfully skips the transaction or
226+
finishes a transaction, LSN (stored in
227+
<structname>pg_subscription</structname>.<structfield>subskiplsn</structfield>)
228+
is cleared. See <xref linkend="logical-replication-conflicts"/> for
229+
the details of logical replication conflicts. Using this command requires
230+
superuser privilege.
231+
</para>
232+
233+
<para>
234+
<replaceable>skip_option</replaceable> specifies options for this operation.
235+
The supported option is:
236+
237+
<variablelist>
238+
<varlistentry>
239+
<term><literal>lsn</literal> (<type>pg_lsn</type>)</term>
240+
<listitem>
241+
<para>
242+
Specifies the finish LSN of the remote transaction whose changes
243+
are to be skipped by the logical replication worker. The finish LSN
244+
is the LSN at which the transaction is either committed or prepared.
245+
Skipping individual subtransaction is not supported. Setting
246+
<literal>NONE</literal> resets the LSN.
247+
</para>
248+
</listitem>
249+
</varlistentry>
250+
</variablelist>
251+
</para>
252+
</listitem>
253+
</varlistentry>
254+
213255
<varlistentry>
214256
<term><replaceable class="parameter">new_owner</replaceable></term>
215257
<listitem>

‎src/backend/catalog/pg_subscription.c

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -70,6 +70,7 @@ GetSubscription(Oid subid, bool missing_ok)
7070
sub->stream=subform->substream;
7171
sub->twophasestate=subform->subtwophasestate;
7272
sub->disableonerr=subform->subdisableonerr;
73+
sub->skiplsn=subform->subskiplsn;
7374

7475
/* Get conninfo */
7576
datum=SysCacheGetAttr(SUBSCRIPTIONOID,

‎src/backend/catalog/system_views.sql

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1261,7 +1261,7 @@ REVOKE ALL ON pg_replication_origin_status FROM public;
12611261
-- All columns of pg_subscription except subconninfo are publicly readable.
12621262
REVOKE ALLON pg_subscriptionFROM public;
12631263
GRANTSELECT (oid, subdbid, subname, subowner, subenabled, subbinary,
1264-
substream, subtwophasestate, subdisableonerr, subslotname,
1264+
substream, subtwophasestate, subdisableonerr,subskiplsn,subslotname,
12651265
subsynccommit, subpublications)
12661266
ON pg_subscription TO public;
12671267

‎src/backend/commands/subscriptioncmds.c

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@
4545
#include"utils/guc.h"
4646
#include"utils/lsyscache.h"
4747
#include"utils/memutils.h"
48+
#include"utils/pg_lsn.h"
4849
#include"utils/syscache.h"
4950

5051
/*
@@ -62,6 +63,7 @@
6263
#defineSUBOPT_STREAMING0x00000100
6364
#defineSUBOPT_TWOPHASE_COMMIT0x00000200
6465
#defineSUBOPT_DISABLE_ON_ERR0x00000400
66+
#defineSUBOPT_LSN0x00000800
6567

6668
/* check if the 'val' has 'bits' set */
6769
#defineIsSet(val,bits) (((val) & (bits)) == (bits))
@@ -84,6 +86,7 @@ typedef struct SubOpts
8486
boolstreaming;
8587
booltwophase;
8688
booldisableonerr;
89+
XLogRecPtrlsn;
8790
}SubOpts;
8891

8992
staticList*fetch_table_list(WalReceiverConn*wrconn,List*publications);
@@ -262,6 +265,33 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
262265
opts->specified_opts |=SUBOPT_DISABLE_ON_ERR;
263266
opts->disableonerr=defGetBoolean(defel);
264267
}
268+
elseif (IsSet(supported_opts,SUBOPT_LSN)&&
269+
strcmp(defel->defname,"lsn")==0)
270+
{
271+
char*lsn_str=defGetString(defel);
272+
XLogRecPtrlsn;
273+
274+
if (IsSet(opts->specified_opts,SUBOPT_LSN))
275+
errorConflictingDefElem(defel,pstate);
276+
277+
/* Setting lsn = NONE is treated as resetting LSN */
278+
if (strcmp(lsn_str,"none")==0)
279+
lsn=InvalidXLogRecPtr;
280+
else
281+
{
282+
/* Parse the argument as LSN */
283+
lsn=DatumGetLSN(DirectFunctionCall1(pg_lsn_in,
284+
CStringGetDatum(lsn_str)));
285+
286+
if (XLogRecPtrIsInvalid(lsn))
287+
ereport(ERROR,
288+
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
289+
errmsg("invalid WAL location (LSN): %s",lsn_str)));
290+
}
291+
292+
opts->specified_opts |=SUBOPT_LSN;
293+
opts->lsn=lsn;
294+
}
265295
else
266296
ereport(ERROR,
267297
(errcode(ERRCODE_SYNTAX_ERROR),
@@ -479,6 +509,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
479509
LOGICALREP_TWOPHASE_STATE_PENDING :
480510
LOGICALREP_TWOPHASE_STATE_DISABLED);
481511
values[Anum_pg_subscription_subdisableonerr-1]=BoolGetDatum(opts.disableonerr);
512+
values[Anum_pg_subscription_subskiplsn-1]=LSNGetDatum(InvalidXLogRecPtr);
482513
values[Anum_pg_subscription_subconninfo-1]=
483514
CStringGetTextDatum(conninfo);
484515
if (opts.slot_name)
@@ -1106,6 +1137,48 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
11061137
break;
11071138
}
11081139

1140+
caseALTER_SUBSCRIPTION_SKIP:
1141+
{
1142+
parse_subscription_options(pstate,stmt->options,SUBOPT_LSN,&opts);
1143+
1144+
/* ALTER SUBSCRIPTION ... SKIP supports only LSN option */
1145+
Assert(IsSet(opts.specified_opts,SUBOPT_LSN));
1146+
1147+
if (!superuser())
1148+
ereport(ERROR,
1149+
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1150+
errmsg("must be superuser to skip transaction")));
1151+
1152+
/*
1153+
* If the user sets subskiplsn, we do a sanity check to make
1154+
* sure that the specified LSN is a probable value.
1155+
*/
1156+
if (!XLogRecPtrIsInvalid(opts.lsn))
1157+
{
1158+
RepOriginIdoriginid;
1159+
charoriginname[NAMEDATALEN];
1160+
XLogRecPtrremote_lsn;
1161+
1162+
snprintf(originname,sizeof(originname),"pg_%u",subid);
1163+
originid=replorigin_by_name(originname, false);
1164+
remote_lsn=replorigin_get_progress(originid, false);
1165+
1166+
/* Check the given LSN is at least a future LSN */
1167+
if (!XLogRecPtrIsInvalid(remote_lsn)&&opts.lsn<remote_lsn)
1168+
ereport(ERROR,
1169+
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1170+
errmsg("skip WAL location (LSN %X/%X) must be greater than origin LSN %X/%X",
1171+
LSN_FORMAT_ARGS(opts.lsn),
1172+
LSN_FORMAT_ARGS(remote_lsn))));
1173+
}
1174+
1175+
values[Anum_pg_subscription_subskiplsn-1]=LSNGetDatum(opts.lsn);
1176+
replaces[Anum_pg_subscription_subskiplsn-1]= true;
1177+
1178+
update_tuple= true;
1179+
break;
1180+
}
1181+
11091182
default:
11101183
elog(ERROR,"unrecognized ALTER SUBSCRIPTION kind %d",
11111184
stmt->kind);

‎src/backend/parser/gram.y

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9983,6 +9983,15 @@ AlterSubscriptionStmt:
99839983
(Node *)makeBoolean(false),@1));
99849984
$$ = (Node *)n;
99859985
}
9986+
|ALTERSUBSCRIPTIONnameSKIPdefinition
9987+
{
9988+
AlterSubscriptionStmt *n =
9989+
makeNode(AlterSubscriptionStmt);
9990+
n->kind = ALTER_SUBSCRIPTION_SKIP;
9991+
n->subname =$3;
9992+
n->options =$5;
9993+
$$ = (Node *)n;
9994+
}
99869995
;
99879996

99889997
/*****************************************************************************

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp