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

Commit89fd41b

Browse files
committed
Fix and improve cache invalidation logic for logical decoding.
There are basically three situations in which logical decoding needsto perform cache invalidation. During/After replaying a transactionwith catalog changes, when skipping a uninteresting transaction thatperformed catalog changes and when erroring out while replaying atransaction. Unfortunately these three cases were all done slightlydifferently - partially because8de3e41, which greatly simplifiesmatters, got committed in the midst of the development of logicaldecoding.The actually problematic case was when logical decoding skippedtransaction commits (and thus processed invalidations). When used viathe SQL interface cache invalidation could access the catalog - bad,because we didn't set up enough state to allow that correctly. It'dnot be hard to setup sufficient state, but the simpler solution is toalways perform cache invalidation outside a valid transaction.Also make the different cache invalidation cases look as similar aspossible, to ease code review.This fixes the assertion failure reported by Antonin Houska in53EE02D9.7040702@gmail.com. The presented testcase has been expandedinto a regression test.Backpatch to 9.4, where logical decoding was introduced.
1 parent5a2c184 commit89fd41b

File tree

4 files changed

+152
-42
lines changed

4 files changed

+152
-42
lines changed

‎contrib/test_decoding/Makefile

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

40-
REGRESSCHECKS=ddl rewrite toast permissions decoding_in_xact binary prepared
40+
REGRESSCHECKS=ddl rewrite toast permissions decoding_in_xactdecoding_into_relbinary prepared
4141

4242
regresscheck: all | submake-regress submake-test_decoding
4343
$(MKDIR_P) regression_output
Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
-- test that we can insert the result of a get_changes call into a
2+
-- logged relation. That's really not a good idea in practical terms,
3+
-- but provides a nice test.
4+
SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding');
5+
?column?
6+
----------
7+
init
8+
(1 row)
9+
10+
-- slot works
11+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
12+
data
13+
------
14+
(0 rows)
15+
16+
-- create some changes
17+
CREATE TABLE somechange(id serial primary key);
18+
INSERT INTO somechange DEFAULT VALUES;
19+
CREATE TABLE changeresult AS
20+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
21+
SELECT * FROM changeresult;
22+
data
23+
------------------------------------------------
24+
BEGIN
25+
table public.somechange: INSERT: id[integer]:1
26+
COMMIT
27+
(3 rows)
28+
29+
INSERT INTO changeresult
30+
SELECT data FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
31+
INSERT INTO changeresult
32+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
33+
SELECT * FROM changeresult;
34+
data
35+
--------------------------------------------------------------------------------------------------------------------------------------------------
36+
BEGIN
37+
table public.somechange: INSERT: id[integer]:1
38+
COMMIT
39+
BEGIN
40+
table public.changeresult: INSERT: data[text]:'BEGIN'
41+
table public.changeresult: INSERT: data[text]:'table public.somechange: INSERT: id[integer]:1'
42+
table public.changeresult: INSERT: data[text]:'COMMIT'
43+
COMMIT
44+
BEGIN
45+
table public.changeresult: INSERT: data[text]:'BEGIN'
46+
table public.changeresult: INSERT: data[text]:'table public.somechange: INSERT: id[integer]:1'
47+
table public.changeresult: INSERT: data[text]:'COMMIT'
48+
COMMIT
49+
BEGIN
50+
table public.changeresult: INSERT: data[text]:'BEGIN'
51+
table public.changeresult: INSERT: data[text]:'table public.changeresult: INSERT: data[text]:''BEGIN'''
52+
table public.changeresult: INSERT: data[text]:'table public.changeresult: INSERT: data[text]:''table public.somechange: INSERT: id[integer]:1'''
53+
table public.changeresult: INSERT: data[text]:'table public.changeresult: INSERT: data[text]:''COMMIT'''
54+
table public.changeresult: INSERT: data[text]:'COMMIT'
55+
COMMIT
56+
(20 rows)
57+
58+
DROP TABLE changeresult;
59+
DROP TABLE somechange;
60+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
61+
data
62+
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
63+
BEGIN
64+
table public.changeresult: INSERT: data[text]:'BEGIN'
65+
table public.changeresult: INSERT: data[text]:'table public.changeresult: INSERT: data[text]:''BEGIN'''
66+
table public.changeresult: INSERT: data[text]:'table public.changeresult: INSERT: data[text]:''table public.somechange: INSERT: id[integer]:1'''
67+
table public.changeresult: INSERT: data[text]:'table public.changeresult: INSERT: data[text]:''COMMIT'''
68+
table public.changeresult: INSERT: data[text]:'COMMIT'
69+
table public.changeresult: INSERT: data[text]:'BEGIN'
70+
table public.changeresult: INSERT: data[text]:'table public.changeresult: INSERT: data[text]:''BEGIN'''
71+
table public.changeresult: INSERT: data[text]:'table public.changeresult: INSERT: data[text]:''table public.changeresult: INSERT: data[text]:''''BEGIN'''''''
72+
table public.changeresult: INSERT: data[text]:'table public.changeresult: INSERT: data[text]:''table public.changeresult: INSERT: data[text]:''''table public.somechange: INSERT: id[integer]:1'''''''
73+
table public.changeresult: INSERT: data[text]:'table public.changeresult: INSERT: data[text]:''table public.changeresult: INSERT: data[text]:''''COMMIT'''''''
74+
table public.changeresult: INSERT: data[text]:'table public.changeresult: INSERT: data[text]:''COMMIT'''
75+
table public.changeresult: INSERT: data[text]:'COMMIT'
76+
COMMIT
77+
(14 rows)
78+
79+
SELECT 'stop' FROM pg_drop_replication_slot('regression_slot');
80+
?column?
81+
----------
82+
stop
83+
(1 row)
84+
Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
-- test that we can insert the result of a get_changes call into a
2+
-- logged relation. That's really not a good idea in practical terms,
3+
-- but provides a nice test.
4+
SELECT'init'FROM pg_create_logical_replication_slot('regression_slot','test_decoding');
5+
6+
-- slot works
7+
SELECT dataFROM pg_logical_slot_get_changes('regression_slot',NULL,NULL,'include-xids','0','skip-empty-xacts','1');
8+
9+
-- create some changes
10+
CREATETABLEsomechange(idserialprimary key);
11+
INSERT INTO somechange DEFAULTVALUES;
12+
13+
CREATETABLEchangeresultAS
14+
SELECT dataFROM pg_logical_slot_get_changes('regression_slot',NULL,NULL,'include-xids','0','skip-empty-xacts','1');
15+
16+
SELECT*FROM changeresult;
17+
18+
INSERT INTO changeresult
19+
SELECT dataFROM pg_logical_slot_peek_changes('regression_slot',NULL,NULL,'include-xids','0','skip-empty-xacts','1');
20+
INSERT INTO changeresult
21+
SELECT dataFROM pg_logical_slot_get_changes('regression_slot',NULL,NULL,'include-xids','0','skip-empty-xacts','1');
22+
23+
SELECT*FROM changeresult;
24+
DROPTABLE changeresult;
25+
DROPTABLE somechange;
26+
SELECT dataFROM pg_logical_slot_get_changes('regression_slot',NULL,NULL,'include-xids','0','skip-empty-xacts','1');
27+
SELECT'stop'FROM pg_drop_replication_slot('regression_slot');

‎src/backend/replication/logical/reorderbuffer.c

Lines changed: 40 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -1264,8 +1264,7 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
12641264

12651265
volatileCommandIdcommand_id=FirstCommandId;
12661266
volatileSnapshotsnapshot_now=NULL;
1267-
volatilebooltxn_started= false;
1268-
volatileboolsubtxn_started= false;
1267+
volatileboolusing_subtxn= false;
12691268

12701269
txn=ReorderBufferTXNByXid(rb,xid, false,NULL,InvalidXLogRecPtr,
12711270
false);
@@ -1305,7 +1304,6 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
13051304

13061305
PG_TRY();
13071306
{
1308-
txn_started= false;
13091307

13101308
/*
13111309
* Decoding needs access to syscaches et al., which in turn use
@@ -1317,16 +1315,12 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
13171315
* When we're called via the SQL SRF there's already a transaction
13181316
* started, so start an explicit subtransaction there.
13191317
*/
1320-
if (IsTransactionOrTransactionBlock())
1321-
{
1318+
using_subtxn=IsTransactionOrTransactionBlock();
1319+
1320+
if (using_subtxn)
13221321
BeginInternalSubTransaction("replay");
1323-
subtxn_started= true;
1324-
}
13251322
else
1326-
{
13271323
StartTransactionCommand();
1328-
txn_started= true;
1329-
}
13301324

13311325
rb->begin(rb,txn);
13321326

@@ -1489,22 +1483,22 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
14891483
elog(ERROR,"output plugin used XID %u",
14901484
GetCurrentTransactionId());
14911485

1492-
/* make sure there's no cache pollution */
1493-
ReorderBufferExecuteInvalidations(rb,txn);
1494-
14951486
/* cleanup */
14961487
TeardownHistoricSnapshot(false);
14971488

14981489
/*
1499-
*Abort subtransaction orthe transaction as a whole has the right
1490+
*Abortingthecurrent (sub-)transaction as a whole has the right
15001491
* semantics. We want all locks acquired in here to be released, not
15011492
* reassigned to the parent and we do not want any database access
15021493
* have persistent effects.
15031494
*/
1504-
if (subtxn_started)
1495+
AbortCurrentTransaction();
1496+
1497+
/* make sure there's no cache pollution */
1498+
ReorderBufferExecuteInvalidations(rb,txn);
1499+
1500+
if (using_subtxn)
15051501
RollbackAndReleaseCurrentSubTransaction();
1506-
elseif (txn_started)
1507-
AbortCurrentTransaction();
15081502

15091503
if (snapshot_now->copied)
15101504
ReorderBufferFreeSnap(rb,snapshot_now);
@@ -1520,20 +1514,21 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
15201514

15211515
TeardownHistoricSnapshot(true);
15221516

1523-
if (snapshot_now->copied)
1524-
ReorderBufferFreeSnap(rb,snapshot_now);
1525-
1526-
if (subtxn_started)
1527-
RollbackAndReleaseCurrentSubTransaction();
1528-
elseif (txn_started)
1529-
AbortCurrentTransaction();
1530-
15311517
/*
1532-
*Invalidations in an aborted transactions aren't allowed to do
1533-
* catalog access, so wedon't need to still have the snapshot setup.
1518+
*Force cache invalidation to happen outside of a valid transaction
1519+
*to preventcatalog access as wejust caught an error.
15341520
*/
1521+
AbortCurrentTransaction();
1522+
1523+
/* make sure there's no cache pollution */
15351524
ReorderBufferExecuteInvalidations(rb,txn);
15361525

1526+
if (using_subtxn)
1527+
RollbackAndReleaseCurrentSubTransaction();
1528+
1529+
if (snapshot_now->copied)
1530+
ReorderBufferFreeSnap(rb,snapshot_now);
1531+
15371532
/* remove potential on-disk data, and deallocate */
15381533
ReorderBufferCleanupTXN(rb,txn);
15391534

@@ -1645,20 +1640,24 @@ ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
16451640
*/
16461641
if (txn->base_snapshot!=NULL&&txn->ninvalidations>0)
16471642
{
1648-
/* setup snapshot to perform the invalidations in */
1649-
SetupHistoricSnapshot(txn->base_snapshot,txn->tuplecid_hash);
1650-
PG_TRY();
1651-
{
1652-
ReorderBufferExecuteInvalidations(rb,txn);
1653-
TeardownHistoricSnapshot(false);
1654-
}
1655-
PG_CATCH();
1656-
{
1657-
/* cleanup */
1658-
TeardownHistoricSnapshot(true);
1659-
PG_RE_THROW();
1660-
}
1661-
PG_END_TRY();
1643+
booluse_subtxn=IsTransactionOrTransactionBlock();
1644+
1645+
if (use_subtxn)
1646+
BeginInternalSubTransaction("replay");
1647+
1648+
/*
1649+
* Force invalidations to happen outside of a valid transaction - that
1650+
* way entries will just be marked as invalid without accessing the
1651+
* catalog. That's advantageous because we don't need to setup the
1652+
* full state necessary for catalog access.
1653+
*/
1654+
if (use_subtxn)
1655+
AbortCurrentTransaction();
1656+
1657+
ReorderBufferExecuteInvalidations(rb,txn);
1658+
1659+
if (use_subtxn)
1660+
RollbackAndReleaseCurrentSubTransaction();
16621661
}
16631662
else
16641663
Assert(txn->ninvalidations==0);

0 commit comments

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