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
Appearance settings

Commit 89fd41b

Browse filesBrowse files
anarazelanarazel
committed
Fix and improve cache invalidation logic for logical decoding.
There are basically three situations in which logical decoding needs to perform cache invalidation. During/After replaying a transaction with catalog changes, when skipping a uninteresting transaction that performed catalog changes and when erroring out while replaying a transaction. Unfortunately these three cases were all done slightly differently - partially because 8de3e41, which greatly simplifies matters, got committed in the midst of the development of logical decoding. The actually problematic case was when logical decoding skipped transaction commits (and thus processed invalidations). When used via the SQL interface cache invalidation could access the catalog - bad, because we didn't set up enough state to allow that correctly. It'd not be hard to setup sufficient state, but the simpler solution is to always perform cache invalidation outside a valid transaction. Also make the different cache invalidation cases look as similar as possible, to ease code review. This fixes the assertion failure reported by Antonin Houska in 53EE02D9.7040702@gmail.com. The presented testcase has been expanded into a regression test. Backpatch to 9.4, where logical decoding was introduced.
1 parent 5a2c184 commit 89fd41b
Copy full SHA for 89fd41b

File tree

Expand file treeCollapse file tree

4 files changed

+152
-42
lines changed
Filter options
Expand file treeCollapse file tree

4 files changed

+152
-42
lines changed

‎contrib/test_decoding/Makefile

Copy file name to clipboardExpand all lines: contrib/test_decoding/Makefile
+1-1Lines 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_xact decoding_into_rel binary prepared
4141

4242
regresscheck: all | submake-regress submake-test_decoding
4343
$(MKDIR_P) regression_output

‎contrib/test_decoding/expected/decoding_into_rel.out

Copy file name to clipboard
+84Lines 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+

‎contrib/test_decoding/sql/decoding_into_rel.sql

Copy file name to clipboard
+27Lines 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 data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
8+
9+
-- create some changes
10+
CREATE TABLE somechange(id serial primary key);
11+
INSERT INTO somechange DEFAULT VALUES;
12+
13+
CREATE TABLE changeresult AS
14+
SELECT data FROM 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 data FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
20+
INSERT INTO changeresult
21+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
22+
23+
SELECT * FROM changeresult;
24+
DROP TABLE changeresult;
25+
DROP TABLE somechange;
26+
SELECT data FROM 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

Copy file name to clipboardExpand all lines: src/backend/replication/logical/reorderbuffer.c
+40-41Lines changed: 40 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -1264,8 +1264,7 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
12641264

12651265
volatile CommandId command_id = FirstCommandId;
12661266
volatile Snapshot snapshot_now = NULL;
1267-
volatile bool txn_started = false;
1268-
volatile bool subtxn_started = false;
1267+
volatile bool using_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 or the transaction as a whole has the right
1490+
* Aborting the current (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-
else if (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-
else if (txn_started)
1529-
AbortCurrentTransaction();
1530-
15311517
/*
1532-
* Invalidations in an aborted transactions aren't allowed to do
1533-
* catalog access, so we don't need to still have the snapshot setup.
1518+
* Force cache invalidation to happen outside of a valid transaction
1519+
* to prevent catalog access as we just 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+
bool use_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 (0)
Morty Proxy This is a proxified and sanitized view of the page, visit original site.