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

Commit5aa2350

Browse files
committed
Introduce replication progress tracking infrastructure.
When implementing a replication solution ontop of logical decoding, tworelated problems exist:* How to safely keep track of replication progress* How to change replication behavior, based on the origin of a row; e.g. to avoid loops in bi-directional replication setupsThe solution to these problems, as implemented here, consist out ofthree parts:1) 'replication origins', which identify nodes in a replication setup.2) 'replication progress tracking', which remembers, for each replication origin, how far replay has progressed in a efficient and crash safe manner.3) The ability to filter out changes performed on the behest of a replication origin during logical decoding; this allows complex replication topologies. E.g. by filtering all replayed changes out.Most of this could also be implemented in "userspace", e.g. by insertingadditional rows contain origin information, but that ends up being muchless efficient and more complicated. We don't want to require variousreplication solutions to reimplement logic for this independently. Theinfrastructure is intended to be generic enough to be reusable.This infrastructure also replaces the 'nodeid' infrastructure of committimestamps. It is intended to provide all the former capabilities,except that there's only 2^16 different origins; but now they integratewith logical decoding. Additionally more functionality is accessible viaSQL. Since the commit timestamp infrastructure has also been introducedin 9.5 (commit73c986a) changing the API is not a problem.For now the number of origins for which the replication progress can betracked simultaneously is determined by the max_replication_slotsGUC. That GUC is not a perfect match to configure this, but theredoesn't seem to be sufficient reason to introduce a separate new one.Bumps both catversion and wal page magic.Author: Andres Freund, with contributions from Petr Jelinek and Craig RingerReviewed-By: Heikki Linnakangas, Petr Jelinek, Robert Haas, Steve SingerDiscussion: 20150216002155.GI15326@awork2.anarazel.de, 20140923182422.GA15776@alap3.anarazel.de, 20131114172632.GE7522@alap2.anarazel.de
1 parentc6e96a2 commit5aa2350

File tree

52 files changed

+2766
-89
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

52 files changed

+2766
-89
lines changed

‎contrib/test_decoding/Makefile

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,8 @@ submake-isolation:
3737
submake-test_decoding:
3838
$(MAKE) -C$(top_builddir)/contrib/test_decoding
3939

40-
REGRESSCHECKS=ddl rewrite toast permissions decoding_in_xact decoding_into_rel binary prepared
40+
REGRESSCHECKS=ddl rewrite toast permissions decoding_in_xact decoding_into_rel\
41+
binary prepared replorigin
4142

4243
regresscheck: all | submake-regress submake-test_decoding temp-install
4344
$(MKDIR_P) regression_output
Lines changed: 141 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,141 @@
1+
-- predictability
2+
SET synchronous_commit = on;
3+
CREATE TABLE origin_tbl(id serial primary key, data text);
4+
CREATE TABLE target_tbl(id serial primary key, data text);
5+
SELECT pg_replication_origin_create('test_decoding: regression_slot');
6+
pg_replication_origin_create
7+
------------------------------
8+
1
9+
(1 row)
10+
11+
-- ensure duplicate creations fail
12+
SELECT pg_replication_origin_create('test_decoding: regression_slot');
13+
ERROR: duplicate key value violates unique constraint "pg_replication_origin_roname_index"
14+
DETAIL: Key (roname)=(test_decoding: regression_slot) already exists.
15+
--ensure deletions work (once)
16+
SELECT pg_replication_origin_create('test_decoding: temp');
17+
pg_replication_origin_create
18+
------------------------------
19+
2
20+
(1 row)
21+
22+
SELECT pg_replication_origin_drop('test_decoding: temp');
23+
pg_replication_origin_drop
24+
----------------------------
25+
26+
(1 row)
27+
28+
SELECT pg_replication_origin_drop('test_decoding: temp');
29+
ERROR: cache lookup failed for replication origin 'test_decoding: temp'
30+
SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding');
31+
?column?
32+
----------
33+
init
34+
(1 row)
35+
36+
-- origin tx
37+
INSERT INTO origin_tbl(data) VALUES ('will be replicated and decoded and decoded again');
38+
INSERT INTO target_tbl(data)
39+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
40+
-- as is normal, the insert into target_tbl shows up
41+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
42+
data
43+
----------------------------------------------------------------------------------------------------------------------------------------------------------------------------
44+
BEGIN
45+
table public.target_tbl: INSERT: id[integer]:1 data[text]:'BEGIN'
46+
table public.target_tbl: INSERT: id[integer]:2 data[text]:'table public.origin_tbl: INSERT: id[integer]:1 data[text]:''will be replicated and decoded and decoded again'''
47+
table public.target_tbl: INSERT: id[integer]:3 data[text]:'COMMIT'
48+
COMMIT
49+
(5 rows)
50+
51+
INSERT INTO origin_tbl(data) VALUES ('will be replicated, but not decoded again');
52+
-- mark session as replaying
53+
SELECT pg_replication_origin_session_setup('test_decoding: regression_slot');
54+
pg_replication_origin_session_setup
55+
-------------------------------------
56+
57+
(1 row)
58+
59+
-- ensure we prevent duplicate setup
60+
SELECT pg_replication_origin_session_setup('test_decoding: regression_slot');
61+
ERROR: cannot setup replication origin when one is already setup
62+
BEGIN;
63+
-- setup transaction origin
64+
SELECT pg_replication_origin_xact_setup('0/aabbccdd', '2013-01-01 00:00');
65+
pg_replication_origin_xact_setup
66+
----------------------------------
67+
68+
(1 row)
69+
70+
INSERT INTO target_tbl(data)
71+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'only-local', '1');
72+
COMMIT;
73+
-- check replication progress for the session is correct
74+
SELECT pg_replication_origin_session_progress(false);
75+
pg_replication_origin_session_progress
76+
----------------------------------------
77+
0/AABBCCDD
78+
(1 row)
79+
80+
SELECT pg_replication_origin_session_progress(true);
81+
pg_replication_origin_session_progress
82+
----------------------------------------
83+
0/AABBCCDD
84+
(1 row)
85+
86+
SELECT pg_replication_origin_session_reset();
87+
pg_replication_origin_session_reset
88+
-------------------------------------
89+
90+
(1 row)
91+
92+
SELECT local_id, external_id, remote_lsn, local_lsn <> '0/0' FROM pg_replication_origin_status;
93+
local_id | external_id | remote_lsn | ?column?
94+
----------+--------------------------------+------------+----------
95+
1 | test_decoding: regression_slot | 0/AABBCCDD | t
96+
(1 row)
97+
98+
-- check replication progress identified by name is correct
99+
SELECT pg_replication_origin_progress('test_decoding: regression_slot', false);
100+
pg_replication_origin_progress
101+
--------------------------------
102+
0/AABBCCDD
103+
(1 row)
104+
105+
SELECT pg_replication_origin_progress('test_decoding: regression_slot', true);
106+
pg_replication_origin_progress
107+
--------------------------------
108+
0/AABBCCDD
109+
(1 row)
110+
111+
-- ensure reset requires previously setup state
112+
SELECT pg_replication_origin_session_reset();
113+
ERROR: no replication origin is configured
114+
-- and magically the replayed xact will be filtered!
115+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'only-local', '1');
116+
data
117+
------
118+
(0 rows)
119+
120+
--but new original changes still show up
121+
INSERT INTO origin_tbl(data) VALUES ('will be replicated');
122+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'only-local', '1');
123+
data
124+
--------------------------------------------------------------------------------
125+
BEGIN
126+
table public.origin_tbl: INSERT: id[integer]:3 data[text]:'will be replicated'
127+
COMMIT
128+
(3 rows)
129+
130+
SELECT pg_drop_replication_slot('regression_slot');
131+
pg_drop_replication_slot
132+
--------------------------
133+
134+
(1 row)
135+
136+
SELECT pg_replication_origin_drop('test_decoding: regression_slot');
137+
pg_replication_origin_drop
138+
----------------------------
139+
140+
(1 row)
141+
Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
-- predictability
2+
SET synchronous_commit=on;
3+
4+
CREATETABLEorigin_tbl(idserialprimary key, datatext);
5+
CREATETABLEtarget_tbl(idserialprimary key, datatext);
6+
7+
SELECT pg_replication_origin_create('test_decoding: regression_slot');
8+
-- ensure duplicate creations fail
9+
SELECT pg_replication_origin_create('test_decoding: regression_slot');
10+
11+
--ensure deletions work (once)
12+
SELECT pg_replication_origin_create('test_decoding: temp');
13+
SELECT pg_replication_origin_drop('test_decoding: temp');
14+
SELECT pg_replication_origin_drop('test_decoding: temp');
15+
16+
SELECT'init'FROM pg_create_logical_replication_slot('regression_slot','test_decoding');
17+
18+
-- origin tx
19+
INSERT INTO origin_tbl(data)VALUES ('will be replicated and decoded and decoded again');
20+
INSERT INTO target_tbl(data)
21+
SELECT dataFROM pg_logical_slot_get_changes('regression_slot',NULL,NULL,'include-xids','0','skip-empty-xacts','1');
22+
23+
-- as is normal, the insert into target_tbl shows up
24+
SELECT dataFROM pg_logical_slot_get_changes('regression_slot',NULL,NULL,'include-xids','0','skip-empty-xacts','1');
25+
26+
INSERT INTO origin_tbl(data)VALUES ('will be replicated, but not decoded again');
27+
28+
-- mark session as replaying
29+
SELECT pg_replication_origin_session_setup('test_decoding: regression_slot');
30+
31+
-- ensure we prevent duplicate setup
32+
SELECT pg_replication_origin_session_setup('test_decoding: regression_slot');
33+
34+
BEGIN;
35+
-- setup transaction origin
36+
SELECT pg_replication_origin_xact_setup('0/aabbccdd','2013-01-01 00:00');
37+
INSERT INTO target_tbl(data)
38+
SELECT dataFROM pg_logical_slot_get_changes('regression_slot',NULL,NULL,'include-xids','0','skip-empty-xacts','1','only-local','1');
39+
COMMIT;
40+
41+
-- check replication progress for the session is correct
42+
SELECT pg_replication_origin_session_progress(false);
43+
SELECT pg_replication_origin_session_progress(true);
44+
45+
SELECT pg_replication_origin_session_reset();
46+
47+
SELECT local_id, external_id, remote_lsn, local_lsn<>'0/0'FROM pg_replication_origin_status;
48+
49+
-- check replication progress identified by name is correct
50+
SELECT pg_replication_origin_progress('test_decoding: regression_slot', false);
51+
SELECT pg_replication_origin_progress('test_decoding: regression_slot', true);
52+
53+
-- ensure reset requires previously setup state
54+
SELECT pg_replication_origin_session_reset();
55+
56+
-- and magically the replayed xact will be filtered!
57+
SELECT dataFROM pg_logical_slot_get_changes('regression_slot',NULL,NULL,'include-xids','0','skip-empty-xacts','1','only-local','1');
58+
59+
--but new original changes still show up
60+
INSERT INTO origin_tbl(data)VALUES ('will be replicated');
61+
SELECT dataFROM pg_logical_slot_get_changes('regression_slot',NULL,NULL,'include-xids','0','skip-empty-xacts','1','only-local','1');
62+
63+
SELECT pg_drop_replication_slot('regression_slot');
64+
SELECT pg_replication_origin_drop('test_decoding: regression_slot');

‎contrib/test_decoding/test_decoding.c

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121

2222
#include"replication/output_plugin.h"
2323
#include"replication/logical.h"
24+
#include"replication/origin.h"
2425

2526
#include"utils/builtins.h"
2627
#include"utils/lsyscache.h"
@@ -43,6 +44,7 @@ typedef struct
4344
boolinclude_timestamp;
4445
boolskip_empty_xacts;
4546
boolxact_wrote_changes;
47+
boolonly_local;
4648
}TestDecodingData;
4749

4850
staticvoidpg_decode_startup(LogicalDecodingContext*ctx,OutputPluginOptions*opt,
@@ -59,6 +61,8 @@ static void pg_decode_commit_txn(LogicalDecodingContext *ctx,
5961
staticvoidpg_decode_change(LogicalDecodingContext*ctx,
6062
ReorderBufferTXN*txn,Relationrel,
6163
ReorderBufferChange*change);
64+
staticboolpg_decode_filter(LogicalDecodingContext*ctx,
65+
RepOriginIdorigin_id);
6266

6367
void
6468
_PG_init(void)
@@ -76,6 +80,7 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
7680
cb->begin_cb=pg_decode_begin_txn;
7781
cb->change_cb=pg_decode_change;
7882
cb->commit_cb=pg_decode_commit_txn;
83+
cb->filter_by_origin_cb=pg_decode_filter;
7984
cb->shutdown_cb=pg_decode_shutdown;
8085
}
8186

@@ -97,6 +102,7 @@ pg_decode_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
97102
data->include_xids= true;
98103
data->include_timestamp= false;
99104
data->skip_empty_xacts= false;
105+
data->only_local= false;
100106

101107
ctx->output_plugin_private=data;
102108

@@ -155,6 +161,17 @@ pg_decode_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
155161
errmsg("could not parse value \"%s\" for parameter \"%s\"",
156162
strVal(elem->arg),elem->defname)));
157163
}
164+
elseif (strcmp(elem->defname,"only-local")==0)
165+
{
166+
167+
if (elem->arg==NULL)
168+
data->only_local= true;
169+
elseif (!parse_bool(strVal(elem->arg),&data->only_local))
170+
ereport(ERROR,
171+
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
172+
errmsg("could not parse value \"%s\" for parameter \"%s\"",
173+
strVal(elem->arg),elem->defname)));
174+
}
158175
else
159176
{
160177
ereport(ERROR,
@@ -223,6 +240,17 @@ pg_decode_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
223240
OutputPluginWrite(ctx, true);
224241
}
225242

243+
staticbool
244+
pg_decode_filter(LogicalDecodingContext*ctx,
245+
RepOriginIdorigin_id)
246+
{
247+
TestDecodingData*data=ctx->output_plugin_private;
248+
249+
if (data->only_local&&origin_id!=InvalidRepOriginId)
250+
return true;
251+
return false;
252+
}
253+
226254
/*
227255
* Print literal `outputstr' already represented as string of type `typid'
228256
* into stringbuf `s'.

0 commit comments

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