public inbox for [email protected]
help / color / mirror / Atom feedFrom: Drouvot, Bertrand <[email protected]>
To: Alvaro Herrera <[email protected]>
To: Andres Freund <[email protected]>
Cc: Ibrar Ahmed <[email protected]>
Cc: Amit Khandekar <[email protected]>
Cc: [email protected]
Cc: tushar <[email protected]>
Cc: [pgdg] Robert Haas <[email protected]>
Cc: Rahila Syed <[email protected]>
Cc: pgsql-hackers <[email protected]>
Subject: Re: Minimal logical decoding on standbys
Date: Wed, 27 Oct 2021 08:55:45 +0200
Message-ID: <[email protected]> (raw)
In-Reply-To: <[email protected]>
References: <[email protected]>
<[email protected]>
<[email protected]>
<[email protected]>
Hi,
On 9/15/21 1:36 PM, Drouvot, Bertrand wrote:
>
> Hi,
>
> On 9/9/21 9:17 AM, Drouvot, Bertrand wrote:
>>
>> Hi Alvaro,
>>
>> On 8/2/21 4:56 PM, Drouvot, Bertrand wrote:
>>> Hi Alvaro,
>>>
>>> On 7/28/21 5:26 PM, Alvaro Herrera wrote:
>>>> On 2021-Jul-27, Drouvot, Bertrand wrote:
>>>>
>>>>> diff --git a/src/backend/utils/cache/lsyscache.c
>>>>> b/src/backend/utils/cache/lsyscache.c
>>>>> +bool
>>>>> +get_rel_logical_catalog(Oid relid)
>>>>> +{
>>>>> + bool res;
>>>>> + Relation rel;
>>>>> +
>>>>> + /* assume previously locked */
>>>>> + rel = table_open(relid, NoLock);
>>>>> + res = RelationIsAccessibleInLogicalDecoding(rel);
>>>>> + table_close(rel, NoLock);
>>>>> +
>>>>> + return res;
>>>>> +}
>>>> So RelationIsAccessibleInLogicalDecoding() does a cheap check for
>>>> wal_level which can be done without opening the table; I think this
>>>> function should be rearranged to avoid doing that when not needed.
>>>
>>> Thanks for looking at it.
>>>
>>>
>>>> Also, putting this function in lsyscache.c seems somewhat wrong since
>>>> it's not merely accessing the system caches ...
>>>>
>>>> I think it would be better to move this elsewhere (relcache.c,
>>>> proto in
>>>> relcache.h, perhaps call it
>>>> RelationIdIsAccessibleInLogicalDecoding) and
>>>> short-circuit for the check that can be done before opening the table.
>>
>> So you have in mind to check for XLogLogicalInfoActive() first, and
>> if true, then open the relation and call
>> RelationIsAccessibleInLogicalDecoding()?
>>
>> If so, then what about also creating a new
>> RelationIsAccessibleWhileLogicalWalLevel() or something like this
>> doing the same as RelationIsAccessibleInLogicalDecoding() but without
>> the XLogLogicalInfoActive() check?
>>
>>>> At least the GiST code appears to be able to call this several
>>>> times per
>>>> vacuum run, so it makes sense to short-circuit it for the fast case.
>>>>
>>>> ... though looking at the GiST code again I wonder if it would be more
>>>> sensible to just stash the table's Relation pointer somewhere in the
>>>> context structs
>>
>> Do you have a "good place" in mind?
>>
> Another rebase attached.
>
> The patch proposal to address Andre's walsender corner cases is still
> a dedicated commit (as i think it may be easier to discuss).
>
Another rebase attached (mainly to fix TAP tests failing due to b3b4d8e68a).
@Andres, the patch file number 6 contains an attempt to fix the
Walsender corner case you pointed out.
@Alvaro, I did not look at your remark yet. Do you have a "good place"
in mind? (related to "just stash the table's Relation pointer somewhere
in the context structs")
Given the size of this patch series, I'm wondering if we could start
committing piece per piece (while still working on the corner cases in
parallel).
That would maximize the amount of coverage it gets in the v15
development cycle.
What do you think?
Thanks
Bertrand
From 748b1104bf2f5acfd9936e486412510d228039cd Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 26 Oct 2021 14:30:22 +0000
Subject: [PATCH v25 1/6] Add info in WAL records in preparation for logical
slot conflict handling.
When a WAL replay on standby indicates that a catalog table tuple is
to be deleted by an xid that is greater than a logical slot's
catalog_xmin, then that means the slot's catalog_xmin conflicts with
the xid, and we need to handle the conflict. While subsequent commits
will do the actual conflict handling, this commit adds a new field
onCatalogTable in such WAL records, that is true for catalog tables,
so as to arrange for conflict handling.
Author: Andres Freund (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
src/backend/access/gist/gist.c | 2 +-
src/backend/access/gist/gistbuild.c | 2 +-
src/backend/access/gist/gistutil.c | 4 ++--
src/backend/access/gist/gistxlog.c | 4 +++-
src/backend/access/hash/hashinsert.c | 1 +
src/backend/access/heap/heapam.c | 4 +++-
src/backend/access/heap/pruneheap.c | 1 +
src/backend/access/heap/visibilitymap.c | 2 +-
src/backend/access/nbtree/nbtpage.c | 12 +++++++++---
src/backend/access/spgist/spgvacuum.c | 8 ++++++++
src/backend/utils/cache/lsyscache.c | 15 +++++++++++++++
src/include/access/gist_private.h | 6 +++---
src/include/access/gistxlog.h | 3 ++-
src/include/access/hash_xlog.h | 1 +
src/include/access/heapam_xlog.h | 5 ++++-
src/include/access/nbtxlog.h | 2 ++
src/include/access/spgxlog.h | 1 +
src/include/utils/lsyscache.h | 1 +
src/include/utils/rel.h | 9 +++++++++
19 files changed, 68 insertions(+), 15 deletions(-)
17.9% src/backend/access/gist/
13.1% src/backend/access/heap/
14.5% src/backend/access/nbtree/
8.9% src/backend/access/spgist/
7.2% src/backend/utils/cache/
19.6% src/include/access/
16.6% src/include/utils/
diff --git a/src/backend/access/gist/gist.c b/src/backend/access/gist/gist.c
index 0683f42c25..b6e6340c3c 100644
--- a/src/backend/access/gist/gist.c
+++ b/src/backend/access/gist/gist.c
@@ -348,7 +348,7 @@ gistplacetopage(Relation rel, Size freespace, GISTSTATE *giststate,
for (; ptr; ptr = ptr->next)
{
/* Allocate new page */
- ptr->buffer = gistNewBuffer(rel);
+ ptr->buffer = gistNewBuffer(heapRel, rel);
GISTInitBuffer(ptr->buffer, (is_leaf) ? F_LEAF : 0);
ptr->page = BufferGetPage(ptr->buffer);
ptr->block.blkno = BufferGetBlockNumber(ptr->buffer);
diff --git a/src/backend/access/gist/gistbuild.c b/src/backend/access/gist/gistbuild.c
index baad28c09f..6d948548c3 100644
--- a/src/backend/access/gist/gistbuild.c
+++ b/src/backend/access/gist/gistbuild.c
@@ -290,7 +290,7 @@ gistbuild(Relation heap, Relation index, IndexInfo *indexInfo)
Page page;
/* initialize the root page */
- buffer = gistNewBuffer(index);
+ buffer = gistNewBuffer(heap, index);
Assert(BufferGetBlockNumber(buffer) == GIST_ROOT_BLKNO);
page = BufferGetPage(buffer);
diff --git a/src/backend/access/gist/gistutil.c b/src/backend/access/gist/gistutil.c
index 43ba03b6eb..1d1e21112c 100644
--- a/src/backend/access/gist/gistutil.c
+++ b/src/backend/access/gist/gistutil.c
@@ -820,7 +820,7 @@ gistcheckpage(Relation rel, Buffer buf)
* Caller is responsible for initializing the page by calling GISTInitBuffer
*/
Buffer
-gistNewBuffer(Relation r)
+gistNewBuffer(Relation heapRel, Relation r)
{
Buffer buffer;
bool needLock;
@@ -864,7 +864,7 @@ gistNewBuffer(Relation r)
* page's deleteXid.
*/
if (XLogStandbyInfoActive() && RelationNeedsWAL(r))
- gistXLogPageReuse(r, blkno, GistPageGetDeleteXid(page));
+ gistXLogPageReuse(heapRel, r, blkno, GistPageGetDeleteXid(page));
return buffer;
}
diff --git a/src/backend/access/gist/gistxlog.c b/src/backend/access/gist/gistxlog.c
index 6464cb9281..46aee6f2a9 100644
--- a/src/backend/access/gist/gistxlog.c
+++ b/src/backend/access/gist/gistxlog.c
@@ -596,7 +596,8 @@ gistXLogAssignLSN(void)
* Write XLOG record about reuse of a deleted page.
*/
void
-gistXLogPageReuse(Relation rel, BlockNumber blkno, FullTransactionId latestRemovedXid)
+gistXLogPageReuse(Relation heapRel, Relation rel,
+ BlockNumber blkno, FullTransactionId latestRemovedXid)
{
gistxlogPageReuse xlrec_reuse;
@@ -607,6 +608,7 @@ gistXLogPageReuse(Relation rel, BlockNumber blkno, FullTransactionId latestRemov
*/
/* XLOG stuff */
+ xlrec_reuse.onCatalogTable = RelationIsAccessibleInLogicalDecoding(heapRel);
xlrec_reuse.node = rel->rd_node;
xlrec_reuse.block = blkno;
xlrec_reuse.latestRemovedFullXid = latestRemovedXid;
diff --git a/src/backend/access/hash/hashinsert.c b/src/backend/access/hash/hashinsert.c
index fe9f0df20b..c2b67a25e1 100644
--- a/src/backend/access/hash/hashinsert.c
+++ b/src/backend/access/hash/hashinsert.c
@@ -398,6 +398,7 @@ _hash_vacuum_one_page(Relation rel, Relation hrel, Buffer metabuf, Buffer buf)
xl_hash_vacuum_one_page xlrec;
XLogRecPtr recptr;
+ xlrec.onCatalogTable = RelationIsAccessibleInLogicalDecoding(hrel);
xlrec.latestRemovedXid = latestRemovedXid;
xlrec.ntuples = ndeletable;
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 2da2be1696..3c45c583cd 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -7945,6 +7945,7 @@ log_heap_freeze(Relation reln, Buffer buffer, TransactionId cutoff_xid,
/* nor when there are no tuples to freeze */
Assert(ntuples > 0);
+ xlrec.onCatalogTable = RelationIsAccessibleInLogicalDecoding(reln);
xlrec.cutoff_xid = cutoff_xid;
xlrec.ntuples = ntuples;
@@ -7975,7 +7976,7 @@ log_heap_freeze(Relation reln, Buffer buffer, TransactionId cutoff_xid,
* heap_buffer, if necessary.
*/
XLogRecPtr
-log_heap_visible(RelFileNode rnode, Buffer heap_buffer, Buffer vm_buffer,
+log_heap_visible(Relation rel, Buffer heap_buffer, Buffer vm_buffer,
TransactionId cutoff_xid, uint8 vmflags)
{
xl_heap_visible xlrec;
@@ -7985,6 +7986,7 @@ log_heap_visible(RelFileNode rnode, Buffer heap_buffer, Buffer vm_buffer,
Assert(BufferIsValid(heap_buffer));
Assert(BufferIsValid(vm_buffer));
+ xlrec.onCatalogTable = RelationIsAccessibleInLogicalDecoding(rel);
xlrec.cutoff_xid = cutoff_xid;
xlrec.flags = vmflags;
XLogBeginInsert();
diff --git a/src/backend/access/heap/pruneheap.c b/src/backend/access/heap/pruneheap.c
index db6912e9fa..b848d0e5a3 100644
--- a/src/backend/access/heap/pruneheap.c
+++ b/src/backend/access/heap/pruneheap.c
@@ -323,6 +323,7 @@ heap_page_prune(Relation relation, Buffer buffer,
xlrec.latestRemovedXid = prstate.latestRemovedXid;
xlrec.nredirected = prstate.nredirected;
xlrec.ndead = prstate.ndead;
+ xlrec.onCatalogTable = RelationIsAccessibleInLogicalDecoding(relation);
XLogBeginInsert();
XLogRegisterData((char *) &xlrec, SizeOfHeapPrune);
diff --git a/src/backend/access/heap/visibilitymap.c b/src/backend/access/heap/visibilitymap.c
index 114fbbdd30..2597636bf0 100644
--- a/src/backend/access/heap/visibilitymap.c
+++ b/src/backend/access/heap/visibilitymap.c
@@ -282,7 +282,7 @@ visibilitymap_set(Relation rel, BlockNumber heapBlk, Buffer heapBuf,
if (XLogRecPtrIsInvalid(recptr))
{
Assert(!InRecovery);
- recptr = log_heap_visible(rel->rd_node, heapBuf, vmBuf,
+ recptr = log_heap_visible(rel, heapBuf, vmBuf,
cutoff_xid, flags);
/*
diff --git a/src/backend/access/nbtree/nbtpage.c b/src/backend/access/nbtree/nbtpage.c
index 5bc7c3616a..3fd59b2a3b 100644
--- a/src/backend/access/nbtree/nbtpage.c
+++ b/src/backend/access/nbtree/nbtpage.c
@@ -36,6 +36,7 @@
#include "utils/memdebug.h"
#include "utils/memutils.h"
#include "utils/snapmgr.h"
+#include "utils/lsyscache.h"
static BTMetaPageData *_bt_getmeta(Relation rel, Buffer metabuf);
static void _bt_log_reuse_page(Relation rel, BlockNumber blkno,
@@ -43,7 +44,8 @@ static void _bt_log_reuse_page(Relation rel, BlockNumber blkno,
static void _bt_delitems_delete(Relation rel, Buffer buf,
TransactionId latestRemovedXid,
OffsetNumber *deletable, int ndeletable,
- BTVacuumPosting *updatable, int nupdatable);
+ BTVacuumPosting *updatable, int nupdatable,
+ Relation heapRel);
static char *_bt_delitems_update(BTVacuumPosting *updatable, int nupdatable,
OffsetNumber *updatedoffsets,
Size *updatedbuflen, bool needswal);
@@ -836,6 +838,7 @@ _bt_log_reuse_page(Relation rel, BlockNumber blkno, FullTransactionId safexid)
*/
/* XLOG stuff */
+ xlrec_reuse.onCatalogTable = get_rel_logical_catalog(rel->rd_index->indrelid);
xlrec_reuse.node = rel->rd_node;
xlrec_reuse.block = blkno;
xlrec_reuse.latestRemovedFullXid = safexid;
@@ -1296,7 +1299,8 @@ _bt_delitems_vacuum(Relation rel, Buffer buf,
static void
_bt_delitems_delete(Relation rel, Buffer buf, TransactionId latestRemovedXid,
OffsetNumber *deletable, int ndeletable,
- BTVacuumPosting *updatable, int nupdatable)
+ BTVacuumPosting *updatable, int nupdatable,
+ Relation heapRel)
{
Page page = BufferGetPage(buf);
BTPageOpaque opaque;
@@ -1358,6 +1362,8 @@ _bt_delitems_delete(Relation rel, Buffer buf, TransactionId latestRemovedXid,
XLogRecPtr recptr;
xl_btree_delete xlrec_delete;
+ xlrec_delete.onCatalogTable =
+ RelationIsAccessibleInLogicalDecoding(heapRel);
xlrec_delete.latestRemovedXid = latestRemovedXid;
xlrec_delete.ndeleted = ndeletable;
xlrec_delete.nupdated = nupdatable;
@@ -1685,7 +1691,7 @@ _bt_delitems_delete_check(Relation rel, Buffer buf, Relation heapRel,
/* Physically delete tuples (or TIDs) using deletable (or updatable) */
_bt_delitems_delete(rel, buf, latestRemovedXid, deletable, ndeletable,
- updatable, nupdatable);
+ updatable, nupdatable, heapRel);
/* be tidy */
for (int i = 0; i < nupdatable; i++)
diff --git a/src/backend/access/spgist/spgvacuum.c b/src/backend/access/spgist/spgvacuum.c
index 76fb0374c4..3186885d14 100644
--- a/src/backend/access/spgist/spgvacuum.c
+++ b/src/backend/access/spgist/spgvacuum.c
@@ -27,6 +27,7 @@
#include "storage/indexfsm.h"
#include "storage/lmgr.h"
#include "utils/snapmgr.h"
+#include "utils/lsyscache.h"
/* Entry in pending-list of TIDs we need to revisit */
@@ -503,6 +504,13 @@ vacuumRedirectAndPlaceholder(Relation index, Buffer buffer)
spgxlogVacuumRedirect xlrec;
GlobalVisState *vistest;
+ /*
+ * There is no chance of endless recursion even when we are doing catalog
+ * acceses here; because, spgist is never used for catalogs. Check
+ * comments in RelationIsAccessibleInLogicalDecoding().
+ */
+ xlrec.onCatalogTable = get_rel_logical_catalog(index->rd_index->indrelid);
+
xlrec.nToPlaceholder = 0;
xlrec.newestRedirectXid = InvalidTransactionId;
diff --git a/src/backend/utils/cache/lsyscache.c b/src/backend/utils/cache/lsyscache.c
index 4ebaa552a2..2fda238870 100644
--- a/src/backend/utils/cache/lsyscache.c
+++ b/src/backend/utils/cache/lsyscache.c
@@ -18,6 +18,7 @@
#include "access/hash.h"
#include "access/htup_details.h"
#include "access/nbtree.h"
+#include "access/table.h"
#include "bootstrap/bootstrap.h"
#include "catalog/namespace.h"
#include "catalog/pg_am.h"
@@ -2062,6 +2063,20 @@ get_rel_persistence(Oid relid)
return result;
}
+bool
+get_rel_logical_catalog(Oid relid)
+{
+ bool res;
+ Relation rel;
+
+ /* assume previously locked */
+ rel = table_open(relid, NoLock);
+ res = RelationIsAccessibleInLogicalDecoding(rel);
+ table_close(rel, NoLock);
+
+ return res;
+}
+
/* ---------- TRANSFORM CACHE ---------- */
diff --git a/src/include/access/gist_private.h b/src/include/access/gist_private.h
index 553d364e2d..a0f4015556 100644
--- a/src/include/access/gist_private.h
+++ b/src/include/access/gist_private.h
@@ -440,8 +440,8 @@ extern XLogRecPtr gistXLogPageDelete(Buffer buffer,
FullTransactionId xid, Buffer parentBuffer,
OffsetNumber downlinkOffset);
-extern void gistXLogPageReuse(Relation rel, BlockNumber blkno,
- FullTransactionId latestRemovedXid);
+extern void gistXLogPageReuse(Relation heapRel, Relation rel,
+ BlockNumber blkno, FullTransactionId latestRemovedXid);
extern XLogRecPtr gistXLogUpdate(Buffer buffer,
OffsetNumber *todelete, int ntodelete,
@@ -485,7 +485,7 @@ extern bool gistproperty(Oid index_oid, int attno,
extern bool gistfitpage(IndexTuple *itvec, int len);
extern bool gistnospace(Page page, IndexTuple *itvec, int len, OffsetNumber todelete, Size freespace);
extern void gistcheckpage(Relation rel, Buffer buf);
-extern Buffer gistNewBuffer(Relation r);
+extern Buffer gistNewBuffer(Relation heapRel, Relation r);
extern bool gistPageRecyclable(Page page);
extern void gistfillbuffer(Page page, IndexTuple *itup, int len,
OffsetNumber off);
diff --git a/src/include/access/gistxlog.h b/src/include/access/gistxlog.h
index fd5144f258..73999ddc70 100644
--- a/src/include/access/gistxlog.h
+++ b/src/include/access/gistxlog.h
@@ -49,9 +49,9 @@ typedef struct gistxlogPageUpdate
*/
typedef struct gistxlogDelete
{
+ bool onCatalogTable;
TransactionId latestRemovedXid;
uint16 ntodelete; /* number of deleted offsets */
-
/*
* In payload of blk 0 : todelete OffsetNumbers
*/
@@ -97,6 +97,7 @@ typedef struct gistxlogPageDelete
*/
typedef struct gistxlogPageReuse
{
+ bool onCatalogTable;
RelFileNode node;
BlockNumber block;
FullTransactionId latestRemovedFullXid;
diff --git a/src/include/access/hash_xlog.h b/src/include/access/hash_xlog.h
index 4353a32dbb..94c3292c1e 100644
--- a/src/include/access/hash_xlog.h
+++ b/src/include/access/hash_xlog.h
@@ -250,6 +250,7 @@ typedef struct xl_hash_init_bitmap_page
*/
typedef struct xl_hash_vacuum_one_page
{
+ bool onCatalogTable;
TransactionId latestRemovedXid;
int ntuples;
diff --git a/src/include/access/heapam_xlog.h b/src/include/access/heapam_xlog.h
index 27db48184e..eba48b0aee 100644
--- a/src/include/access/heapam_xlog.h
+++ b/src/include/access/heapam_xlog.h
@@ -242,6 +242,7 @@ typedef struct xl_heap_update
*/
typedef struct xl_heap_prune
{
+ bool onCatalogTable;
TransactionId latestRemovedXid;
uint16 nredirected;
uint16 ndead;
@@ -338,6 +339,7 @@ typedef struct xl_heap_freeze_tuple
*/
typedef struct xl_heap_freeze_page
{
+ bool onCatalogTable;
TransactionId cutoff_xid;
uint16 ntuples;
} xl_heap_freeze_page;
@@ -352,6 +354,7 @@ typedef struct xl_heap_freeze_page
*/
typedef struct xl_heap_visible
{
+ bool onCatalogTable;
TransactionId cutoff_xid;
uint8 flags;
} xl_heap_visible;
@@ -413,7 +416,7 @@ extern bool heap_prepare_freeze_tuple(HeapTupleHeader tuple,
bool *totally_frozen);
extern void heap_execute_freeze_tuple(HeapTupleHeader tuple,
xl_heap_freeze_tuple *xlrec_tp);
-extern XLogRecPtr log_heap_visible(RelFileNode rnode, Buffer heap_buffer,
+extern XLogRecPtr log_heap_visible(Relation rel, Buffer heap_buffer,
Buffer vm_buffer, TransactionId cutoff_xid, uint8 flags);
#endif /* HEAPAM_XLOG_H */
diff --git a/src/include/access/nbtxlog.h b/src/include/access/nbtxlog.h
index 0f7731856b..b15aa47f1b 100644
--- a/src/include/access/nbtxlog.h
+++ b/src/include/access/nbtxlog.h
@@ -185,6 +185,7 @@ typedef struct xl_btree_dedup
*/
typedef struct xl_btree_reuse_page
{
+ bool onCatalogTable;
RelFileNode node;
BlockNumber block;
FullTransactionId latestRemovedFullXid;
@@ -232,6 +233,7 @@ typedef struct xl_btree_vacuum
typedef struct xl_btree_delete
{
+ bool onCatalogTable;
TransactionId latestRemovedXid;
uint16 ndeleted;
uint16 nupdated;
diff --git a/src/include/access/spgxlog.h b/src/include/access/spgxlog.h
index 69405b5750..06b91f4d04 100644
--- a/src/include/access/spgxlog.h
+++ b/src/include/access/spgxlog.h
@@ -237,6 +237,7 @@ typedef struct spgxlogVacuumRoot
typedef struct spgxlogVacuumRedirect
{
+ bool onCatalogTable;
uint16 nToPlaceholder; /* number of redirects to make placeholders */
OffsetNumber firstPlaceholder; /* first placeholder tuple to remove */
TransactionId newestRedirectXid; /* newest XID of removed redirects */
diff --git a/src/include/utils/lsyscache.h b/src/include/utils/lsyscache.h
index 77871aaefc..e2a5efed30 100644
--- a/src/include/utils/lsyscache.h
+++ b/src/include/utils/lsyscache.h
@@ -139,6 +139,7 @@ extern char get_rel_relkind(Oid relid);
extern bool get_rel_relispartition(Oid relid);
extern Oid get_rel_tablespace(Oid relid);
extern char get_rel_persistence(Oid relid);
+extern bool get_rel_logical_catalog(Oid relid);
extern Oid get_transform_fromsql(Oid typid, Oid langid, List *trftypes);
extern Oid get_transform_tosql(Oid typid, Oid langid, List *trftypes);
extern bool get_typisdefined(Oid typid);
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index b4faa1c123..648aeacd78 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -27,6 +27,7 @@
#include "storage/smgr.h"
#include "utils/relcache.h"
#include "utils/reltrigger.h"
+#include "catalog/catalog.h"
/*
@@ -365,6 +366,9 @@ typedef struct StdRdOptions
* RelationIsUsedAsCatalogTable
* Returns whether the relation should be treated as a catalog table
* from the pov of logical decoding. Note multiple eval of argument!
+ * This definition should not invoke anything that performs catalog
+ * access; otherwise it may cause infinite recursion. Check the comments
+ * in RelationIsAccessibleInLogicalDecoding() for details.
*/
#define RelationIsUsedAsCatalogTable(relation) \
((relation)->rd_options && \
@@ -653,6 +657,11 @@ RelationGetSmgr(Relation rel)
* RelationIsAccessibleInLogicalDecoding
* True if we need to log enough information to have access via
* decoding snapshot.
+ * This definition should not invoke anything that performs catalog
+ * access. Otherwise, e.g. logging a WAL entry for catalog relation may
+ * invoke this function, which will in turn do catalog access, which may
+ * in turn cause another similar WAL entry to be logged, leading to
+ * infinite recursion.
*/
#define RelationIsAccessibleInLogicalDecoding(relation) \
(XLogLogicalInfoActive() && \
--
2.18.4
From 24b0541619a2a176e502120c3bb1c8e7d3ad2276 Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 26 Oct 2021 14:31:14 +0000
Subject: [PATCH v25 2/6] Handle logical slot conflicts on standby.
During WAL replay on standby, when slot conflict is identified,
invalidate such slots. Also do the same thing if wal_level on master
is reduced to below logical and there are existing logical slots
on standby. Introduce a new ProcSignalReason value for slot
conflict recovery. Arrange for a new pg_stat_get_activity field:
confl_logicalslot.
Author: Andres Freund (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
doc/src/sgml/monitoring.sgml | 10 +
src/backend/access/gist/gistxlog.c | 4 +-
src/backend/access/hash/hash_xlog.c | 3 +-
src/backend/access/heap/heapam.c | 10 +-
src/backend/access/nbtree/nbtxlog.c | 4 +-
src/backend/access/spgist/spgxlog.c | 1 +
src/backend/access/transam/xlog.c | 13 ++
src/backend/catalog/system_views.sql | 3 +-
src/backend/postmaster/pgstat.c | 20 ++
.../replication/logical/logicalfuncs.c | 7 +-
src/backend/replication/slot.c | 209 ++++++++++++++++++
src/backend/replication/walsender.c | 8 +
src/backend/storage/ipc/procarray.c | 4 +
src/backend/storage/ipc/procsignal.c | 3 +
src/backend/storage/ipc/standby.c | 11 +-
src/backend/tcop/postgres.c | 22 ++
src/backend/utils/adt/pgstatfuncs.c | 16 ++
src/include/catalog/pg_proc.dat | 5 +
src/include/pgstat.h | 2 +
src/include/replication/slot.h | 2 +
src/include/storage/procsignal.h | 1 +
src/include/storage/standby.h | 4 +-
src/test/regress/expected/rules.out | 3 +-
23 files changed, 350 insertions(+), 15 deletions(-)
4.1% src/backend/access/heap/
4.7% src/backend/access/transam/
5.1% src/backend/access/
4.3% src/backend/postmaster/
3.5% src/backend/replication/logical/
52.2% src/backend/replication/
5.6% src/backend/storage/ipc/
7.1% src/backend/tcop/
3.6% src/backend/
6.3% src/include/
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 3173ec2566..bd6723bd63 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -4018,6 +4018,16 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
deadlocks
</para></entry>
</row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>confl_logicalslot</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of queries in this database that have been canceled due to
+ logical slots
+ </para></entry>
+ </row>
</tbody>
</tgroup>
</table>
diff --git a/src/backend/access/gist/gistxlog.c b/src/backend/access/gist/gistxlog.c
index 46aee6f2a9..5963e639d8 100644
--- a/src/backend/access/gist/gistxlog.c
+++ b/src/backend/access/gist/gistxlog.c
@@ -195,7 +195,8 @@ gistRedoDeleteRecord(XLogReaderState *record)
XLogRecGetBlockTag(record, 0, &rnode, NULL, NULL);
- ResolveRecoveryConflictWithSnapshot(xldata->latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xldata->latestRemovedXid,
+ xldata->onCatalogTable, rnode);
}
if (XLogReadBufferForRedo(record, 0, &buffer) == BLK_NEEDS_REDO)
@@ -395,6 +396,7 @@ gistRedoPageReuse(XLogReaderState *record)
*/
if (InHotStandby)
ResolveRecoveryConflictWithSnapshotFullXid(xlrec->latestRemovedFullXid,
+ xlrec->onCatalogTable,
xlrec->node);
}
diff --git a/src/backend/access/hash/hash_xlog.c b/src/backend/access/hash/hash_xlog.c
index 27475fcbd6..e5c6124400 100644
--- a/src/backend/access/hash/hash_xlog.c
+++ b/src/backend/access/hash/hash_xlog.c
@@ -1002,7 +1002,8 @@ hash_xlog_vacuum_one_page(XLogReaderState *record)
RelFileNode rnode;
XLogRecGetBlockTag(record, 0, &rnode, NULL, NULL);
- ResolveRecoveryConflictWithSnapshot(xldata->latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xldata->latestRemovedXid,
+ xldata->onCatalogTable, rnode);
}
action = XLogReadBufferForRedoExtended(record, 0, RBM_NORMAL, true, &buffer);
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 3c45c583cd..43624603a0 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -8423,7 +8423,8 @@ heap_xlog_prune(XLogReaderState *record)
* no queries running for which the removed tuples are still visible.
*/
if (InHotStandby)
- ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid,
+ xlrec->onCatalogTable, rnode);
/*
* If we have a full-page image, restore it (using a cleanup lock) and
@@ -8591,7 +8592,9 @@ heap_xlog_visible(XLogReaderState *record)
* rather than killing the transaction outright.
*/
if (InHotStandby)
- ResolveRecoveryConflictWithSnapshot(xlrec->cutoff_xid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xlrec->cutoff_xid,
+ xlrec->onCatalogTable,
+ rnode);
/*
* Read the heap page, if it still exists. If the heap file has dropped or
@@ -8728,7 +8731,8 @@ heap_xlog_freeze_page(XLogReaderState *record)
TransactionIdRetreat(latestRemovedXid);
XLogRecGetBlockTag(record, 0, &rnode, NULL, NULL);
- ResolveRecoveryConflictWithSnapshot(latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(latestRemovedXid,
+ xlrec->onCatalogTable, rnode);
}
if (XLogReadBufferForRedo(record, 0, &buffer) == BLK_NEEDS_REDO)
diff --git a/src/backend/access/nbtree/nbtxlog.c b/src/backend/access/nbtree/nbtxlog.c
index 786c08c0ce..41b7ec8e2d 100644
--- a/src/backend/access/nbtree/nbtxlog.c
+++ b/src/backend/access/nbtree/nbtxlog.c
@@ -668,7 +668,8 @@ btree_xlog_delete(XLogReaderState *record)
XLogRecGetBlockTag(record, 0, &rnode, NULL, NULL);
- ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid,
+ xlrec->onCatalogTable, rnode);
}
/*
@@ -1006,6 +1007,7 @@ btree_xlog_reuse_page(XLogReaderState *record)
if (InHotStandby)
ResolveRecoveryConflictWithSnapshotFullXid(xlrec->latestRemovedFullXid,
+ xlrec->onCatalogTable,
xlrec->node);
}
diff --git a/src/backend/access/spgist/spgxlog.c b/src/backend/access/spgist/spgxlog.c
index 3dfd2aa317..add4da4e74 100644
--- a/src/backend/access/spgist/spgxlog.c
+++ b/src/backend/access/spgist/spgxlog.c
@@ -881,6 +881,7 @@ spgRedoVacuumRedirect(XLogReaderState *record)
XLogRecGetBlockTag(record, 0, &node, NULL, NULL);
ResolveRecoveryConflictWithSnapshot(xldata->newestRedirectXid,
+ xldata->onCatalogTable,
node);
}
}
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index f547efd294..0d2590ebeb 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -10577,6 +10577,19 @@ xlog_redo(XLogReaderState *record)
/* Update our copy of the parameters in pg_control */
memcpy(&xlrec, XLogRecGetData(record), sizeof(xl_parameter_change));
+ /*
+ * Invalidate logical slots if we are in hot standby and the primary does not
+ * have a WAL level sufficient for logical decoding. No need to search
+ * for potentially conflicting logically slots if standby is running
+ * with wal_level lower than logical, because in that case, we would
+ * have either disallowed creation of logical slots or invalidated existing
+ * ones.
+ */
+ if (InRecovery && InHotStandby &&
+ xlrec.wal_level < WAL_LEVEL_LOGICAL &&
+ wal_level >= WAL_LEVEL_LOGICAL)
+ InvalidateConflictingLogicalReplicationSlots(InvalidOid,InvalidTransactionId);
+
LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
ControlFile->MaxConnections = xlrec.MaxConnections;
ControlFile->max_worker_processes = xlrec.max_worker_processes;
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 55f6e3711d..cd1fc88d17 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1020,7 +1020,8 @@ CREATE VIEW pg_stat_database_conflicts AS
pg_stat_get_db_conflict_lock(D.oid) AS confl_lock,
pg_stat_get_db_conflict_snapshot(D.oid) AS confl_snapshot,
pg_stat_get_db_conflict_bufferpin(D.oid) AS confl_bufferpin,
- pg_stat_get_db_conflict_startup_deadlock(D.oid) AS confl_deadlock
+ pg_stat_get_db_conflict_startup_deadlock(D.oid) AS confl_deadlock,
+ pg_stat_get_db_conflict_logicalslot(D.oid) AS confl_logicalslot
FROM pg_database D;
CREATE VIEW pg_stat_user_functions AS
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index b7d0fbaefd..4fad4a9b1c 100644
--- a/src/backend/postmaster/pgstat.c
+++ b/src/backend/postmaster/pgstat.c
@@ -1869,6 +1869,22 @@ pgstat_report_replslot_drop(const char *slotname)
pgstat_send(&msg, sizeof(PgStat_MsgReplSlot));
}
+/* ----------
+ * pgstat_report_replslot_conflict()
+ * Tell the collector about a logical slot being conflicting
+ * with recovery.
+ * ----------
+ */
+void
+pgstat_report_replslot_conflict(Oid dboid)
+{
+ PgStat_MsgRecoveryConflict msg;
+ pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_RECOVERYCONFLICT);
+ msg.m_databaseid = dboid;
+ msg.m_reason = PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT;
+ pgstat_send(&msg, sizeof(msg));
+}
+
/* ----------
* pgstat_ping() -
*
@@ -3633,6 +3649,7 @@ reset_dbentry_counters(PgStat_StatDBEntry *dbentry)
dbentry->n_conflict_tablespace = 0;
dbentry->n_conflict_lock = 0;
dbentry->n_conflict_snapshot = 0;
+ dbentry->n_conflict_logicalslot = 0;
dbentry->n_conflict_bufferpin = 0;
dbentry->n_conflict_startup_deadlock = 0;
dbentry->n_temp_files = 0;
@@ -5580,6 +5597,9 @@ pgstat_recv_recoveryconflict(PgStat_MsgRecoveryConflict *msg, int len)
case PROCSIG_RECOVERY_CONFLICT_SNAPSHOT:
dbentry->n_conflict_snapshot++;
break;
+ case PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT:
+ dbentry->n_conflict_logicalslot++;
+ break;
case PROCSIG_RECOVERY_CONFLICT_BUFFERPIN:
dbentry->n_conflict_bufferpin++;
break;
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index e59939aad1..805d4c5a5b 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -232,11 +232,14 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
/*
* After the sanity checks in CreateDecodingContext, make sure the
- * restart_lsn is valid. Avoid "cannot get changes" wording in this
+ * restart_lsn is valid or both xmin and catalog_xmin are valid.
+ * Avoid "cannot get changes" wording in this
* errmsg because that'd be confusingly ambiguous about no changes
* being available.
*/
- if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn))
+ if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn)
+ || (!TransactionIdIsValid(MyReplicationSlot->data.xmin)
+ && !TransactionIdIsValid(MyReplicationSlot->data.catalog_xmin)))
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("can no longer get changes from replication slot \"%s\"",
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 1c6c0c7ce2..188716957f 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -1350,6 +1350,215 @@ restart:
return invalidated;
}
+/*
+ * Helper for InvalidateConflictingLogicalReplicationSlot -- acquires the given slot
+ * and mark it invalid, if necessary and possible.
+ *
+ * Returns whether ReplicationSlotControlLock was released in the interim (and
+ * in that case we're not holding the lock at return, otherwise we are).
+ *
+ * This is inherently racy, because we release the LWLock
+ * for syscalls, so caller must restart if we return true.
+ */
+static bool
+InvalidatePossiblyConflictingLogicalReplicationSlot(ReplicationSlot *s, TransactionId xid)
+{
+ int last_signaled_pid = 0;
+ bool released_lock = false;
+
+ for (;;)
+ {
+ TransactionId slot_xmin;
+ TransactionId slot_catalog_xmin;
+ NameData slotname;
+ int active_pid = 0;
+
+ Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
+
+ if (!s->in_use)
+ {
+ if (released_lock)
+ LWLockRelease(ReplicationSlotControlLock);
+ break;
+ }
+
+ /*
+ * Check if the slot needs to be invalidated. If it needs to be
+ * invalidated, and is not currently acquired, acquire it and mark it
+ * as having been invalidated. We do this with the spinlock held to
+ * avoid race conditions -- for example the xmin(s) could move forward
+ * , or the slot could be dropped.
+ */
+ SpinLockAcquire(&s->mutex);
+
+ slot_xmin = s->data.xmin;
+ slot_catalog_xmin = s->data.catalog_xmin;
+
+ /*
+ * If the slot is already invalid or is not conflicting, we don't need to
+ * do anything.
+ */
+
+ /* slot has been invalidated */
+ if ((!TransactionIdIsValid(slot_xmin) && !TransactionIdIsValid(slot_catalog_xmin))
+ ||
+ /*
+ * we are not forcing for invalidation because the xid is valid
+ * and this is a non conflicting slot
+ */
+ (TransactionIdIsValid(xid) && !(
+ (TransactionIdIsValid(slot_xmin) && TransactionIdPrecedesOrEquals(slot_xmin, xid))
+ ||
+ (TransactionIdIsValid(slot_catalog_xmin) && TransactionIdPrecedesOrEquals(slot_catalog_xmin, xid))
+ ))
+ )
+ {
+ SpinLockRelease(&s->mutex);
+ if (released_lock)
+ LWLockRelease(ReplicationSlotControlLock);
+ break;
+ }
+
+ slotname = s->data.name;
+ active_pid = s->active_pid;
+
+ /*
+ * If the slot can be acquired, do so and mark it invalidated
+ * immediately. Otherwise we'll signal the owning process, below, and
+ * retry.
+ */
+ if (active_pid == 0)
+ {
+ MyReplicationSlot = s;
+ s->active_pid = MyProcPid;
+ s->data.xmin = InvalidTransactionId;
+ s->data.catalog_xmin = InvalidTransactionId;
+ }
+
+ SpinLockRelease(&s->mutex);
+
+ if (active_pid != 0)
+ {
+ /*
+ * Prepare the sleep on the slot's condition variable before
+ * releasing the lock, to close a possible race condition if the
+ * slot is released before the sleep below.
+ */
+
+ ConditionVariablePrepareToSleep(&s->active_cv);
+
+ LWLockRelease(ReplicationSlotControlLock);
+ released_lock = true;
+
+ /*
+ * Signal to terminate the process that owns the slot, if we
+ * haven't already signalled it. (Avoidance of repeated
+ * signalling is the only reason for there to be a loop in this
+ * routine; otherwise we could rely on caller's restart loop.)
+ *
+ * There is the race condition that other process may own the slot
+ * after its current owner process is terminated and before this
+ * process owns it. To handle that, we signal only if the PID of
+ * the owning process has changed from the previous time. (This
+ * logic assumes that the same PID is not reused very quickly.)
+ */
+ if (last_signaled_pid != active_pid)
+ {
+ ereport(LOG,
+ (errmsg("terminating process %d because replication slot \"%s\" conflicts with recovery",
+ active_pid, NameStr(slotname))));
+
+ (void) kill(active_pid, SIGTERM);
+ last_signaled_pid = active_pid;
+ }
+
+ /* Wait until the slot is released. */
+ ConditionVariableSleep(&s->active_cv,
+ WAIT_EVENT_REPLICATION_SLOT_DROP);
+
+ /*
+ * Re-acquire lock and start over; we expect to invalidate the
+ * slot next time (unless another process acquires the slot in the
+ * meantime).
+ */
+ LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+ continue;
+ }
+ else
+ {
+ /*
+ * We hold the slot now and have already invalidated it; flush it
+ * to ensure that state persists.
+ *
+ * Don't want to hold ReplicationSlotControlLock across file
+ * system operations, so release it now but be sure to tell caller
+ * to restart from scratch.
+ */
+ LWLockRelease(ReplicationSlotControlLock);
+ released_lock = true;
+
+ /* Make sure the invalidated state persists across server restart */
+ ReplicationSlotMarkDirty();
+ ReplicationSlotSave();
+ ReplicationSlotRelease();
+ pgstat_report_replslot_conflict(s->data.database);
+
+ ereport(LOG,
+ (errmsg("invalidating slot \"%s\" because it conflicts with recovery", NameStr(slotname))));
+
+ /* done with this slot for now */
+ break;
+ }
+ }
+
+ Assert(!released_lock == LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
+
+ return released_lock;
+}
+
+/*
+ * Resolve recovery conflicts with logical slots.
+ *
+ * When xid is valid, it means that we are about to remove rows older than xid.
+ * Therefore we need to invalidate slots that depend on seeing those rows.
+ * When xid is invalid, invalidate all logical slots. This is required when the
+ * master wal_level is set back to replica, so existing logical slots need to
+ * be invalidated.
+ */
+void
+InvalidateConflictingLogicalReplicationSlots(Oid dboid, TransactionId xid)
+{
+
+ Assert(max_replication_slots >= 0);
+
+ if (max_replication_slots == 0)
+ return;
+restart:
+ LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+ for (int i = 0; i < max_replication_slots; i++)
+ {
+ ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
+
+ if (!s->in_use)
+ continue;
+
+ /* We are only dealing with *logical* slot conflicts. */
+ if (!SlotIsLogical(s))
+ continue;
+
+ /* not our database and we don't want all the database, skip */
+ if (s->data.database != dboid && TransactionIdIsValid(xid))
+ continue;
+
+ if (InvalidatePossiblyConflictingLogicalReplicationSlot(s, xid))
+ {
+ /* if the lock was released, we need to restart from scratch */
+ goto restart;
+ }
+ }
+ LWLockRelease(ReplicationSlotControlLock);
+}
+
/*
* Flush all replication slots to disk.
*
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index d9ab6d6de2..63c0dd2f62 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1246,6 +1246,14 @@ StartLogicalReplication(StartReplicationCmd *cmd)
ReplicationSlotAcquire(cmd->slotname, true);
+ if (!TransactionIdIsValid(MyReplicationSlot->data.xmin)
+ && !TransactionIdIsValid(MyReplicationSlot->data.catalog_xmin))
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot read from logical replication slot \"%s\"",
+ cmd->slotname),
+ errdetail("This slot has been invalidated because it was conflicting with recovery.")));
+
if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn))
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index bd3c7a47fe..23ffc39a0a 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -3443,6 +3443,10 @@ SignalVirtualTransaction(VirtualTransactionId vxid, ProcSignalReason sigmode,
GET_VXID_FROM_PGPROC(procvxid, *proc);
+ /*
+ * Note: vxid.localTransactionId can be invalid, which means the
+ * request is to signal the pid that is not running a transaction.
+ */
if (procvxid.backendId == vxid.backendId &&
procvxid.localTransactionId == vxid.localTransactionId)
{
diff --git a/src/backend/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c
index defb75aa26..315d5a1e33 100644
--- a/src/backend/storage/ipc/procsignal.c
+++ b/src/backend/storage/ipc/procsignal.c
@@ -673,6 +673,9 @@ procsignal_sigusr1_handler(SIGNAL_ARGS)
if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_SNAPSHOT))
RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_SNAPSHOT);
+ if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT))
+ RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT);
+
if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK))
RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK);
diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c
index b17326bc20..111fec1b9f 100644
--- a/src/backend/storage/ipc/standby.c
+++ b/src/backend/storage/ipc/standby.c
@@ -34,6 +34,7 @@
#include "utils/ps_status.h"
#include "utils/timeout.h"
#include "utils/timestamp.h"
+#include "replication/slot.h"
/* User-settable GUC parameters */
int vacuum_defer_cleanup_age;
@@ -440,7 +441,8 @@ ResolveRecoveryConflictWithVirtualXIDs(VirtualTransactionId *waitlist,
}
void
-ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid, RelFileNode node)
+ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid,
+ bool onCatalogTable, RelFileNode node)
{
VirtualTransactionId *backends;
@@ -465,6 +467,9 @@ ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid, RelFileNode
PROCSIG_RECOVERY_CONFLICT_SNAPSHOT,
WAIT_EVENT_RECOVERY_CONFLICT_SNAPSHOT,
true);
+
+ if (onCatalogTable)
+ InvalidateConflictingLogicalReplicationSlots(node.dbNode, latestRemovedXid);
}
/*
@@ -473,7 +478,7 @@ ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid, RelFileNode
*/
void
ResolveRecoveryConflictWithSnapshotFullXid(FullTransactionId latestRemovedFullXid,
- RelFileNode node)
+ bool onCatalogTable, RelFileNode node)
{
/*
* ResolveRecoveryConflictWithSnapshot operates on 32-bit TransactionIds,
@@ -491,7 +496,7 @@ ResolveRecoveryConflictWithSnapshotFullXid(FullTransactionId latestRemovedFullXi
TransactionId latestRemovedXid;
latestRemovedXid = XidFromFullTransactionId(latestRemovedFullXid);
- ResolveRecoveryConflictWithSnapshot(latestRemovedXid, node);
+ ResolveRecoveryConflictWithSnapshot(latestRemovedXid, onCatalogTable, node);
}
}
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index 0775abe35d..697b0ea7ad 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -2467,6 +2467,9 @@ errdetail_recovery_conflict(void)
case PROCSIG_RECOVERY_CONFLICT_SNAPSHOT:
errdetail("User query might have needed to see row versions that must be removed.");
break;
+ case PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT:
+ errdetail("User was using the logical slot that must be dropped.");
+ break;
case PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK:
errdetail("User transaction caused buffer deadlock with recovery.");
break;
@@ -3037,6 +3040,25 @@ RecoveryConflictInterrupt(ProcSignalReason reason)
case PROCSIG_RECOVERY_CONFLICT_LOCK:
case PROCSIG_RECOVERY_CONFLICT_TABLESPACE:
case PROCSIG_RECOVERY_CONFLICT_SNAPSHOT:
+ case PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT:
+ /*
+ * For conflicts that require a logical slot to be invalidated, the
+ * requirement is for the signal receiver to release the slot,
+ * so that it could be invalidated by the signal sender. So for
+ * normal backends, the transaction should be aborted, just
+ * like for other recovery conflicts. But if it's walsender on
+ * standby, then it has to be killed so as to release an
+ * acquired logical slot.
+ */
+ if (am_cascading_walsender &&
+ reason == PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT &&
+ MyReplicationSlot && SlotIsLogical(MyReplicationSlot))
+ {
+ RecoveryConflictPending = true;
+ QueryCancelPending = true;
+ InterruptPending = true;
+ break;
+ }
/*
* If we aren't in a transaction any longer then ignore.
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index ff5aedc99c..073b402bac 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -1497,6 +1497,21 @@ pg_stat_get_db_conflict_snapshot(PG_FUNCTION_ARGS)
PG_RETURN_INT64(result);
}
+Datum
+pg_stat_get_db_conflict_logicalslot(PG_FUNCTION_ARGS)
+{
+ Oid dbid = PG_GETARG_OID(0);
+ int64 result;
+ PgStat_StatDBEntry *dbentry;
+
+ if ((dbentry = pgstat_fetch_stat_dbentry(dbid)) == NULL)
+ result = 0;
+ else
+ result = (int64) (dbentry->n_conflict_logicalslot);
+
+ PG_RETURN_INT64(result);
+}
+
Datum
pg_stat_get_db_conflict_bufferpin(PG_FUNCTION_ARGS)
{
@@ -1540,6 +1555,7 @@ pg_stat_get_db_conflict_all(PG_FUNCTION_ARGS)
result = (int64) (dbentry->n_conflict_tablespace +
dbentry->n_conflict_lock +
dbentry->n_conflict_snapshot +
+ dbentry->n_conflict_logicalslot +
dbentry->n_conflict_bufferpin +
dbentry->n_conflict_startup_deadlock);
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index d068d6532e..e40f57a549 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5502,6 +5502,11 @@
proname => 'pg_stat_get_db_conflict_snapshot', provolatile => 's',
proparallel => 'r', prorettype => 'int8', proargtypes => 'oid',
prosrc => 'pg_stat_get_db_conflict_snapshot' },
+{ oid => '4544',
+ descr => 'statistics: recovery conflicts in database caused by logical replication slot',
+ proname => 'pg_stat_get_db_conflict_logicalslot', provolatile => 's',
+ proparallel => 'r', prorettype => 'int8', proargtypes => 'oid',
+ prosrc => 'pg_stat_get_db_conflict_logicalslot' },
{ oid => '3068',
descr => 'statistics: recovery conflicts in database caused by shared buffer pin',
proname => 'pg_stat_get_db_conflict_bufferpin', provolatile => 's',
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index bcd3588ea2..3570e4933a 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -747,6 +747,7 @@ typedef struct PgStat_StatDBEntry
PgStat_Counter n_conflict_tablespace;
PgStat_Counter n_conflict_lock;
PgStat_Counter n_conflict_snapshot;
+ PgStat_Counter n_conflict_logicalslot;
PgStat_Counter n_conflict_bufferpin;
PgStat_Counter n_conflict_startup_deadlock;
PgStat_Counter n_temp_files;
@@ -1038,6 +1039,7 @@ extern void pgstat_report_checksum_failure(void);
extern void pgstat_report_replslot(const PgStat_StatReplSlotEntry *repSlotStat);
extern void pgstat_report_replslot_create(const char *slotname);
extern void pgstat_report_replslot_drop(const char *slotname);
+extern void pgstat_report_replslot_conflict(Oid dbOid);
extern void pgstat_initialize(void);
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 53d773ccff..cca4bb058c 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -214,6 +214,7 @@ extern XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN(void);
extern bool ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive);
extern void ReplicationSlotsDropDBSlots(Oid dboid);
extern bool InvalidateObsoleteReplicationSlots(XLogSegNo oldestSegno);
+extern void InvalidateConflictingLogicalReplicationSlots(Oid dboid, TransactionId xid);
extern ReplicationSlot *SearchNamedReplicationSlot(const char *name, bool need_lock);
extern void ReplicationSlotNameForTablesync(Oid suboid, Oid relid, char *syncslotname, int szslot);
extern void ReplicationSlotDropAtPubNode(WalReceiverConn *wrconn, char *slotname, bool missing_ok);
@@ -223,5 +224,6 @@ extern void CheckPointReplicationSlots(void);
extern void CheckSlotRequirements(void);
extern void CheckSlotPermissions(void);
+extern void ResolveRecoveryConflictWithLogicalSlots(Oid dboid, TransactionId xid, char *reason);
#endif /* SLOT_H */
diff --git a/src/include/storage/procsignal.h b/src/include/storage/procsignal.h
index eec186be2e..f86b070dbc 100644
--- a/src/include/storage/procsignal.h
+++ b/src/include/storage/procsignal.h
@@ -41,6 +41,7 @@ typedef enum
PROCSIG_RECOVERY_CONFLICT_TABLESPACE,
PROCSIG_RECOVERY_CONFLICT_LOCK,
PROCSIG_RECOVERY_CONFLICT_SNAPSHOT,
+ PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT,
PROCSIG_RECOVERY_CONFLICT_BUFFERPIN,
PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK,
diff --git a/src/include/storage/standby.h b/src/include/storage/standby.h
index 38fd85a431..3ba1882216 100644
--- a/src/include/storage/standby.h
+++ b/src/include/storage/standby.h
@@ -30,9 +30,9 @@ extern void InitRecoveryTransactionEnvironment(void);
extern void ShutdownRecoveryTransactionEnvironment(void);
extern void ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid,
- RelFileNode node);
+ bool onCatalogTable, RelFileNode node);
extern void ResolveRecoveryConflictWithSnapshotFullXid(FullTransactionId latestRemovedFullXid,
- RelFileNode node);
+ bool onCatalogTable, RelFileNode node);
extern void ResolveRecoveryConflictWithTablespace(Oid tsid);
extern void ResolveRecoveryConflictWithDatabase(Oid dbid);
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 2fa00a3c29..7cf918634c 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1871,7 +1871,8 @@ pg_stat_database_conflicts| SELECT d.oid AS datid,
pg_stat_get_db_conflict_lock(d.oid) AS confl_lock,
pg_stat_get_db_conflict_snapshot(d.oid) AS confl_snapshot,
pg_stat_get_db_conflict_bufferpin(d.oid) AS confl_bufferpin,
- pg_stat_get_db_conflict_startup_deadlock(d.oid) AS confl_deadlock
+ pg_stat_get_db_conflict_startup_deadlock(d.oid) AS confl_deadlock,
+ pg_stat_get_db_conflict_logicalslot(d.oid) AS confl_logicalslot
FROM pg_database d;
pg_stat_gssapi| SELECT s.pid,
s.gss_auth AS gss_authenticated,
--
2.18.4
From 178a6d1c8fecf13164a5b65a8d5677b92dc40aa6 Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 26 Oct 2021 14:32:11 +0000
Subject: [PATCH v25 3/6] Allow logical decoding on standby.
Allow a logical slot to be created on standby. Restrict its usage
or its creation if wal_level on primary is less than logical.
During slot creation, it's restart_lsn is set to the last replayed
LSN. Effectively, a logical slot creation on standby waits for an
xl_running_xact record to arrive from primary. Conflicting slots
would be handled in next commits.
Author: Andres Freund (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
src/backend/access/transam/xlog.c | 33 ++++++++++-
src/backend/access/transam/xlogfuncs.c | 2 +-
src/backend/access/transam/xlogutils.c | 2 +-
src/backend/postmaster/checkpointer.c | 4 +-
src/backend/replication/logical/decode.c | 22 +++++++-
src/backend/replication/logical/logical.c | 37 ++++++------
.../replication/logical/logicalfuncs.c | 2 +-
src/backend/replication/slot.c | 56 ++++++++++---------
src/backend/replication/slotfuncs.c | 2 +-
src/backend/replication/walreceiver.c | 4 +-
src/backend/replication/walreceiverfuncs.c | 2 +-
src/backend/replication/walsender.c | 28 +++++++---
src/include/access/xlog.h | 3 +-
13 files changed, 133 insertions(+), 64 deletions(-)
16.4% src/backend/access/transam/
32.2% src/backend/replication/logical/
46.0% src/backend/replication/
5.2% src/
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 0d2590ebeb..1e9f94927c 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -5123,6 +5123,17 @@ LocalProcessControlFile(bool reset)
ReadControlFile();
}
+/*
+ * Get the wal_level from the control file. For a standby, this value should be
+ * considered as its active wal_level, because it may be different from what
+ * was originally configured on standby.
+ */
+WalLevel
+GetActiveWalLevelOnStandby(void)
+{
+ return ControlFile->wal_level;
+}
+
/*
* Initialization of shared memory for XLOG
*/
@@ -9787,7 +9798,7 @@ CreateRestartPoint(int flags)
* whichever is later.
*/
receivePtr = GetWalRcvFlushRecPtr(NULL, NULL);
- replayPtr = GetXLogReplayRecPtr(&replayTLI);
+ replayPtr = GetXLogReplayRecPtr(&replayTLI, false);
endptr = (receivePtr < replayPtr) ? replayPtr : receivePtr;
KeepLogSeg(endptr, &_logSegNo);
if (InvalidateObsoleteReplicationSlots(_logSegNo))
@@ -11952,7 +11963,7 @@ register_persistent_abort_backup_handler(void)
* Exported to allow WALReceiver to read the pointer directly.
*/
XLogRecPtr
-GetXLogReplayRecPtr(TimeLineID *replayTLI)
+GetXLogReplayRecPtr(TimeLineID *replayTLI, bool avoid_header)
{
XLogRecPtr recptr;
TimeLineID tli;
@@ -11964,6 +11975,24 @@ GetXLogReplayRecPtr(TimeLineID *replayTLI)
if (replayTLI)
*replayTLI = tli;
+
+ /*
+ * Replay pointer may point one past the end of the record. If that
+ * is a XLOG page boundary, it will not be a valid LSN for the
+ * start of a record, so bump it up past the page header.
+ */
+ if (avoid_header && !XRecOffIsValid(recptr))
+ {
+ if (recptr % XLOG_BLCKSZ != 0)
+ elog(ERROR, "invalid replay pointer");
+
+ /* For the first page of a segment file, it's a long header */
+ if (XLogSegmentOffset(recptr, wal_segment_size) == 0)
+ recptr += SizeOfXLogLongPHD;
+ else
+ recptr += SizeOfXLogShortPHD;
+ }
+
return recptr;
}
diff --git a/src/backend/access/transam/xlogfuncs.c b/src/backend/access/transam/xlogfuncs.c
index b98deb72ec..a173f8d6fc 100644
--- a/src/backend/access/transam/xlogfuncs.c
+++ b/src/backend/access/transam/xlogfuncs.c
@@ -417,7 +417,7 @@ pg_last_wal_replay_lsn(PG_FUNCTION_ARGS)
{
XLogRecPtr recptr;
- recptr = GetXLogReplayRecPtr(NULL);
+ recptr = GetXLogReplayRecPtr(NULL, false);
if (recptr == 0)
PG_RETURN_NULL();
diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c
index 88a1bfd939..aaade9382d 100644
--- a/src/backend/access/transam/xlogutils.c
+++ b/src/backend/access/transam/xlogutils.c
@@ -870,7 +870,7 @@ read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
if (!RecoveryInProgress())
read_upto = GetFlushRecPtr();
else
- read_upto = GetXLogReplayRecPtr(&ThisTimeLineID);
+ read_upto = GetXLogReplayRecPtr(&ThisTimeLineID, false);
tli = ThisTimeLineID;
/*
diff --git a/src/backend/postmaster/checkpointer.c b/src/backend/postmaster/checkpointer.c
index be7366379d..ea1bf7d247 100644
--- a/src/backend/postmaster/checkpointer.c
+++ b/src/backend/postmaster/checkpointer.c
@@ -434,7 +434,7 @@ CheckpointerMain(void)
*/
ckpt_active = true;
if (do_restartpoint)
- ckpt_start_recptr = GetXLogReplayRecPtr(NULL);
+ ckpt_start_recptr = GetXLogReplayRecPtr(NULL, false);
else
ckpt_start_recptr = GetInsertRecPtr();
ckpt_start_time = now;
@@ -794,7 +794,7 @@ IsCheckpointOnSchedule(double progress)
* value that was in effect when the WAL was generated).
*/
if (RecoveryInProgress())
- recptr = GetXLogReplayRecPtr(NULL);
+ recptr = GetXLogReplayRecPtr(NULL, false);
else
recptr = GetInsertRecPtr();
elapsed_xlogs = (((double) (recptr - ckpt_start_recptr)) /
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index 2874dc0612..b8be0c83ed 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -215,11 +215,31 @@ DecodeXLogOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
* can restart from there.
*/
break;
+ case XLOG_PARAMETER_CHANGE:
+ {
+ xl_parameter_change *xlrec =
+ (xl_parameter_change *) XLogRecGetData(buf->record);
+
+ /*
+ * If wal_level on primary is reduced to less than logical, then we
+ * want to prevent existing logical slots from being used.
+ * Existing logical slots on standby get invalidated when this WAL
+ * record is replayed; and further, slot creation fails when the
+ * wal level is not sufficient; but all these operations are not
+ * synchronized, so a logical slot may creep in while the wal_level
+ * is being reduced. Hence this extra check.
+ */
+ if (xlrec->wal_level < WAL_LEVEL_LOGICAL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("logical decoding on standby requires "
+ "wal_level >= logical on master")));
+ break;
+ }
case XLOG_NOOP:
case XLOG_NEXTOID:
case XLOG_SWITCH:
case XLOG_BACKUP_END:
- case XLOG_PARAMETER_CHANGE:
case XLOG_RESTORE_POINT:
case XLOG_FPW_CHANGE:
case XLOG_FPI_FOR_HINT:
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index aae0ae5b8a..1e8b2808d5 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -119,23 +119,22 @@ CheckLogicalDecodingRequirements(void)
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("logical decoding requires a database connection")));
- /* ----
- * TODO: We got to change that someday soon...
- *
- * There's basically three things missing to allow this:
- * 1) We need to be able to correctly and quickly identify the timeline a
- * LSN belongs to
- * 2) We need to force hot_standby_feedback to be enabled at all times so
- * the primary cannot remove rows we need.
- * 3) support dropping replication slots referring to a database, in
- * dbase_redo. There can't be any active ones due to HS recovery
- * conflicts, so that should be relatively easy.
- * ----
- */
if (RecoveryInProgress())
- ereport(ERROR,
- (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
- errmsg("logical decoding cannot be used while in recovery")));
+ {
+ /*
+ * This check may have race conditions, but whenever
+ * XLOG_PARAMETER_CHANGE indicates that wal_level has changed, we
+ * verify that there are no existing logical replication slots. And to
+ * avoid races around creating a new slot,
+ * CheckLogicalDecodingRequirements() is called once before creating
+ * the slot, and once when logical decoding is initially starting up.
+ */
+ if (GetActiveWalLevelOnStandby() < WAL_LEVEL_LOGICAL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("logical decoding on standby requires "
+ "wal_level >= logical on master")));
+ }
}
/*
@@ -330,6 +329,12 @@ CreateInitDecodingContext(const char *plugin,
LogicalDecodingContext *ctx;
MemoryContext old_context;
+ /*
+ * On standby, this check is also required while creating the slot. Check
+ * the comments in this function.
+ */
+ CheckLogicalDecodingRequirements();
+
/* shorter lines... */
slot = MyReplicationSlot;
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index 805d4c5a5b..7d8890a22a 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -214,7 +214,7 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
if (!RecoveryInProgress())
end_of_wal = GetFlushRecPtr();
else
- end_of_wal = GetXLogReplayRecPtr(&ThisTimeLineID);
+ end_of_wal = GetXLogReplayRecPtr(&ThisTimeLineID, false);
ReplicationSlotAcquire(NameStr(*name), true);
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 188716957f..81a3e72732 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -1100,37 +1100,28 @@ ReplicationSlotReserveWal(void)
/*
* For logical slots log a standby snapshot and start logical decoding
* at exactly that position. That allows the slot to start up more
- * quickly.
+ * quickly. But on a standby we cannot do WAL writes, so just use the
+ * replay pointer; effectively, an attempt to create a logical slot on
+ * standby will cause it to wait for an xl_running_xact record to be
+ * logged independently on the primary, so that a snapshot can be built
+ * using the record.
*
- * That's not needed (or indeed helpful) for physical slots as they'll
- * start replay at the last logged checkpoint anyway. Instead return
- * the location of the last redo LSN. While that slightly increases
- * the chance that we have to retry, it's where a base backup has to
- * start replay at.
+ * None of this is needed (or indeed helpful) for physical slots as
+ * they'll start replay at the last logged checkpoint anyway. Instead
+ * return the location of the last redo LSN. While that slightly
+ * increases the chance that we have to retry, it's where a base backup
+ * has to start replay at.
*/
- if (!RecoveryInProgress() && SlotIsLogical(slot))
- {
- XLogRecPtr flushptr;
-
- /* start at current insert position */
+ if (SlotIsPhysical(slot))
+ restart_lsn = GetRedoRecPtr();
+ else if (RecoveryInProgress())
+ restart_lsn = GetXLogReplayRecPtr(NULL, true);
+ else
restart_lsn = GetXLogInsertRecPtr();
- SpinLockAcquire(&slot->mutex);
- slot->data.restart_lsn = restart_lsn;
- SpinLockRelease(&slot->mutex);
-
- /* make sure we have enough information to start */
- flushptr = LogStandbySnapshot();
- /* and make sure it's fsynced to disk */
- XLogFlush(flushptr);
- }
- else
- {
- restart_lsn = GetRedoRecPtr();
- SpinLockAcquire(&slot->mutex);
- slot->data.restart_lsn = restart_lsn;
- SpinLockRelease(&slot->mutex);
- }
+ SpinLockAcquire(&slot->mutex);
+ slot->data.restart_lsn = restart_lsn;
+ SpinLockRelease(&slot->mutex);
/* prevent WAL removal as fast as possible */
ReplicationSlotsComputeRequiredLSN();
@@ -1146,6 +1137,17 @@ ReplicationSlotReserveWal(void)
if (XLogGetLastRemovedSegno() < segno)
break;
}
+
+ if (!RecoveryInProgress() && SlotIsLogical(slot))
+ {
+ XLogRecPtr flushptr;
+
+ /* make sure we have enough information to start */
+ flushptr = LogStandbySnapshot();
+
+ /* and make sure it's fsynced to disk */
+ XLogFlush(flushptr);
+ }
}
/*
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 17df99c2ac..948acb1ccb 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -627,7 +627,7 @@ pg_replication_slot_advance(PG_FUNCTION_ARGS)
if (!RecoveryInProgress())
moveto = Min(moveto, GetFlushRecPtr());
else
- moveto = Min(moveto, GetXLogReplayRecPtr(&ThisTimeLineID));
+ moveto = Min(moveto, GetXLogReplayRecPtr(&ThisTimeLineID, false));
/* Acquire the slot so we "own" it */
ReplicationSlotAcquire(NameStr(*slotname), true);
diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index b90e5ca98e..28cc123c09 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -408,7 +408,7 @@ WalReceiverMain(void)
first_stream = false;
/* Initialize LogstreamResult and buffers for processing messages */
- LogstreamResult.Write = LogstreamResult.Flush = GetXLogReplayRecPtr(NULL);
+ LogstreamResult.Write = LogstreamResult.Flush = GetXLogReplayRecPtr(NULL, false);
initStringInfo(&reply_message);
initStringInfo(&incoming_message);
@@ -1098,7 +1098,7 @@ XLogWalRcvSendReply(bool force, bool requestReply)
/* Construct a new message */
writePtr = LogstreamResult.Write;
flushPtr = LogstreamResult.Flush;
- applyPtr = GetXLogReplayRecPtr(NULL);
+ applyPtr = GetXLogReplayRecPtr(NULL, false);
resetStringInfo(&reply_message);
pq_sendbyte(&reply_message, 'r');
diff --git a/src/backend/replication/walreceiverfuncs.c b/src/backend/replication/walreceiverfuncs.c
index 6f0acbfdef..f264b71f73 100644
--- a/src/backend/replication/walreceiverfuncs.c
+++ b/src/backend/replication/walreceiverfuncs.c
@@ -372,7 +372,7 @@ GetReplicationApplyDelay(void)
receivePtr = walrcv->flushedUpto;
SpinLockRelease(&walrcv->mutex);
- replayPtr = GetXLogReplayRecPtr(NULL);
+ replayPtr = GetXLogReplayRecPtr(NULL, false);
if (receivePtr == replayPtr)
return 0;
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 63c0dd2f62..2ea4d27b63 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -535,7 +535,7 @@ ReadReplicationSlot(ReadReplicationSlotCmd *cmd)
* to get the LSN position's history.
*/
if (RecoveryInProgress())
- (void) GetXLogReplayRecPtr(¤t_timeline);
+ (void) GetXLogReplayRecPtr(¤t_timeline, false);
else
current_timeline = ThisTimeLineID;
@@ -1273,6 +1273,16 @@ StartLogicalReplication(StartReplicationCmd *cmd)
got_STOPPING = true;
}
+ /*
+ * In case of logical decoding on standby it may be that ThisTimeLineID
+ * is not set yet.
+ * Indeed we are not going through InitXLOGAccess on a Standby and
+ * it may also be that IdentifySystem has not been called yet.
+ * So let's get it through GetXLogReplayRecPtr().
+ */
+ if (ThisTimeLineID == 0)
+ (void) GetXLogReplayRecPtr(&ThisTimeLineID, false);
+
/*
* Create our decoding context, making it start at the previously ack'ed
* position.
@@ -1497,7 +1507,7 @@ WalSndWaitForWal(XLogRecPtr loc)
if (!RecoveryInProgress())
RecentFlushPtr = GetFlushRecPtr();
else
- RecentFlushPtr = GetXLogReplayRecPtr(NULL);
+ RecentFlushPtr = GetXLogReplayRecPtr(NULL, false);
for (;;)
{
@@ -1531,7 +1541,7 @@ WalSndWaitForWal(XLogRecPtr loc)
if (!RecoveryInProgress())
RecentFlushPtr = GetFlushRecPtr();
else
- RecentFlushPtr = GetXLogReplayRecPtr(NULL);
+ RecentFlushPtr = GetXLogReplayRecPtr(NULL, false);
/*
* If postmaster asked us to stop, don't wait anymore.
@@ -3004,10 +3014,12 @@ XLogSendLogical(void)
* If first time through in this session, initialize flushPtr. Otherwise,
* we only need to update flushPtr if EndRecPtr is past it.
*/
- if (flushPtr == InvalidXLogRecPtr)
- flushPtr = GetFlushRecPtr();
- else if (logical_decoding_ctx->reader->EndRecPtr >= flushPtr)
- flushPtr = GetFlushRecPtr();
+ if (flushPtr == InvalidXLogRecPtr ||
+ logical_decoding_ctx->reader->EndRecPtr >= flushPtr)
+ {
+ flushPtr = (am_cascading_walsender ?
+ GetStandbyFlushRecPtr() : GetFlushRecPtr());
+ }
/* If EndRecPtr is still past our flushPtr, it means we caught up. */
if (logical_decoding_ctx->reader->EndRecPtr >= flushPtr)
@@ -3096,7 +3108,7 @@ GetStandbyFlushRecPtr(void)
*/
receivePtr = GetWalRcvFlushRecPtr(NULL, &receiveTLI);
- replayPtr = GetXLogReplayRecPtr(&replayTLI);
+ replayPtr = GetXLogReplayRecPtr(&replayTLI, false);
ThisTimeLineID = replayTLI;
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 5e2c94a05f..97061253e6 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -282,7 +282,7 @@ extern bool HotStandbyActive(void);
extern bool HotStandbyActiveInReplay(void);
extern bool XLogInsertAllowed(void);
extern void GetXLogReceiptTime(TimestampTz *rtime, bool *fromStream);
-extern XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI);
+extern XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI, bool avoid_header);
extern XLogRecPtr GetXLogInsertRecPtr(void);
extern XLogRecPtr GetXLogWriteRecPtr(void);
extern RecoveryPauseState GetRecoveryPauseState(void);
@@ -299,6 +299,7 @@ extern Size XLOGShmemSize(void);
extern void XLOGShmemInit(void);
extern void BootStrapXLOG(void);
extern void LocalProcessControlFile(bool reset);
+extern WalLevel GetActiveWalLevelOnStandby(void);
extern void StartupXLOG(void);
extern void ShutdownXLOG(int code, Datum arg);
extern void InitXLOGAccess(void);
--
2.18.4
From 873882b8dad1969124016785bc61243a658632a7 Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 26 Oct 2021 14:33:52 +0000
Subject: [PATCH v25 4/6] New TAP test for logical decoding on standby.
Author: Craig Ringer (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
src/test/perl/PostgreSQL/Test/Cluster.pm | 37 ++
.../t/027_standby_logical_decoding.pl | 498 ++++++++++++++++++
2 files changed, 535 insertions(+)
6.0% src/test/perl/PostgreSQL/Test/
93.9% src/test/recovery/t/
diff --git a/src/test/perl/PostgreSQL/Test/Cluster.pm b/src/test/perl/PostgreSQL/Test/Cluster.pm
index 86eb920ea1..89e49facb4 100644
--- a/src/test/perl/PostgreSQL/Test/Cluster.pm
+++ b/src/test/perl/PostgreSQL/Test/Cluster.pm
@@ -2804,6 +2804,43 @@ sub pg_recvlogical_upto
=pod
+=item $node->create_logical_slot_on_standby(self, master, slot_name, dbname)
+
+Create logical replication slot on given standby
+
+=cut
+
+sub create_logical_slot_on_standby
+{
+ my ($self, $master, $slot_name, $dbname) = @_;
+ my ($stdout, $stderr);
+
+ my $handle;
+
+ $handle = IPC::Run::start(['pg_recvlogical', '-d', $self->connstr($dbname), '-P', 'test_decoding', '-S', $slot_name, '--create-slot'], '>', \$stdout, '2>', \$stderr);
+
+ # Once slot restart_lsn is created, the standby looks for xl_running_xacts
+ # WAL record from the restart_lsn onwards. So firstly, wait until the slot
+ # restart_lsn is evaluated.
+
+ $self->poll_query_until(
+ 'postgres', qq[
+ SELECT restart_lsn IS NOT NULL
+ FROM pg_catalog.pg_replication_slots WHERE slot_name = '$slot_name'
+ ]) or die "timed out waiting for logical slot to calculate its restart_lsn";
+
+ # Now arrange for the xl_running_xacts record for which pg_recvlogical
+ # is waiting.
+ $master->safe_psql('postgres', 'CHECKPOINT');
+
+ $handle->finish();
+
+ is($self->slot($slot_name)->{'slot_type'}, 'logical', $slot_name . ' on standby created')
+ or die "could not create slot" . $slot_name;
+}
+
+=pod
+
=back
=cut
diff --git a/src/test/recovery/t/027_standby_logical_decoding.pl b/src/test/recovery/t/027_standby_logical_decoding.pl
new file mode 100644
index 0000000000..cb75f2b0c3
--- /dev/null
+++ b/src/test/recovery/t/027_standby_logical_decoding.pl
@@ -0,0 +1,498 @@
+# logical decoding on standby : test logical decoding,
+# recovery conflict and standby promotion.
+
+use strict;
+use warnings;
+
+use PostgreSQL::Test::Cluster;
+use Test::More tests => 38;
+use Time::HiRes qw(usleep);
+
+my ($stdin, $stdout, $stderr, $ret, $handle, $slot);
+
+my $node_primary = PostgreSQL::Test::Cluster->new('primary');
+my $node_standby = PostgreSQL::Test::Cluster->new('standby');
+
+# Name for the physical slot on primary
+my $primary_slotname = 'primary_physical';
+
+# return the size of logfile of $node in bytes
+sub get_log_size
+{
+ my ($node) = @_;
+
+ return (stat $node->logfile)[7];
+}
+
+# find $pat in logfile of $node after $off-th byte
+sub find_in_log
+{
+ my ($node, $pat, $off) = @_;
+
+ $off = 0 unless defined $off;
+ my $log = PostgreSQL::Test::Utils::slurp_file($node->logfile);
+ return 0 if (length($log) <= $off);
+
+ $log = substr($log, $off);
+
+ return $log =~ m/$pat/;
+}
+
+# Fetch xmin columns from slot's pg_replication_slots row, after waiting for
+# given boolean condition to be true to ensure we've reached a quiescent state.
+sub wait_for_xmins
+{
+ my ($node, $slotname, $check_expr) = @_;
+
+ $node->poll_query_until(
+ 'postgres', qq[
+ SELECT $check_expr
+ FROM pg_catalog.pg_replication_slots
+ WHERE slot_name = '$slotname';
+ ]) or die "Timed out waiting for slot xmins to advance";
+}
+
+# Create the required logical slots on standby.
+sub create_logical_slots
+{
+ $node_standby->create_logical_slot_on_standby($node_primary, 'inactiveslot', 'testdb');
+ $node_standby->create_logical_slot_on_standby($node_primary, 'activeslot', 'testdb');
+}
+
+# Acquire one of the standby logical slots created by create_logical_slots().
+# In case wait is true we are waiting for an active pid on the 'activeslot' slot.
+# If wait is not true it means we are testing a known failure scenario.
+sub make_slot_active
+{
+ my $wait = shift;
+ my $slot_user_handle;
+
+ print "starting pg_recvlogical\n";
+ $slot_user_handle = IPC::Run::start(['pg_recvlogical', '-d', $node_standby->connstr('testdb'), '-S', 'activeslot', '-f', '-', '--no-loop', '--start'], '>', \$stdout, '2>', \$stderr);
+
+ if ($wait)
+ # make sure activeslot is in use
+ {
+ $node_standby->poll_query_until('testdb',
+ "SELECT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = 'activeslot' AND active_pid IS NOT NULL)"
+ ) or die "slot never became active";
+ }
+
+ return $slot_user_handle;
+}
+
+# Check if all the slots on standby are dropped. These include the 'activeslot'
+# that was acquired by make_slot_active(), and the non-active 'inactiveslot'.
+sub check_slots_dropped
+{
+ my ($slot_user_handle) = @_;
+ my $return;
+
+ is($node_standby->slot('inactiveslot')->{'slot_type'}, '', 'inactiveslot on standby dropped');
+ is($node_standby->slot('activeslot')->{'slot_type'}, '', 'activeslot on standby dropped');
+
+ # our client should've terminated in response to the walsender error
+ $slot_user_handle->finish;
+ $return = $?;
+ cmp_ok($return, "!=", 0, "pg_recvlogical exited non-zero");
+ if ($return) {
+ like($stderr, qr/conflict with recovery/, 'slot have been dropped');
+ }
+
+ return 0;
+}
+
+
+########################
+# Initialize primary node
+########################
+
+$node_primary->init(allows_streaming => 1, has_archiving => 1);
+$node_primary->append_conf('postgresql.conf', q{
+wal_level = 'logical'
+max_replication_slots = 4
+max_wal_senders = 4
+log_min_messages = 'debug2'
+log_error_verbosity = verbose
+});
+$node_primary->dump_info;
+$node_primary->start;
+
+$node_primary->psql('postgres', q[CREATE DATABASE testdb]);
+
+$node_primary->safe_psql('testdb', qq[SELECT * FROM pg_create_physical_replication_slot('$primary_slotname');]);
+my $backup_name = 'b1';
+$node_primary->backup($backup_name);
+
+#######################
+# Initialize standby node
+#######################
+
+$node_standby->init_from_backup(
+ $node_primary, $backup_name,
+ has_streaming => 1,
+ has_restoring => 1);
+$node_standby->append_conf('postgresql.conf',
+ qq[primary_slot_name = '$primary_slotname']);
+$node_standby->start;
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+
+##################################################
+# Test that logical decoding on the standby
+# behaves correctly.
+##################################################
+
+create_logical_slots();
+
+$node_primary->safe_psql('testdb', qq[CREATE TABLE decoding_test(x integer, y text);]);
+$node_primary->safe_psql('testdb', qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(1,10) s;]);
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+my $result = $node_standby->safe_psql('testdb',
+ qq[SELECT pg_logical_slot_get_changes('activeslot', NULL, NULL);]);
+
+# test if basic decoding works
+is(scalar(my @foobar = split /^/m, $result),
+ 14, 'Decoding produced 14 rows');
+
+# Insert some rows and verify that we get the same results from pg_recvlogical
+# and the SQL interface.
+$node_primary->safe_psql('testdb',
+ qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(1,4) s;]
+);
+
+my $expected = q{BEGIN
+table public.decoding_test: INSERT: x[integer]:1 y[text]:'1'
+table public.decoding_test: INSERT: x[integer]:2 y[text]:'2'
+table public.decoding_test: INSERT: x[integer]:3 y[text]:'3'
+table public.decoding_test: INSERT: x[integer]:4 y[text]:'4'
+COMMIT};
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+my $stdout_sql = $node_standby->safe_psql('testdb',
+ qq[SELECT data FROM pg_logical_slot_peek_changes('activeslot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');]
+);
+
+is($stdout_sql, $expected, 'got expected output from SQL decoding session');
+
+my $endpos = $node_standby->safe_psql('testdb',
+ "SELECT lsn FROM pg_logical_slot_peek_changes('activeslot', NULL, NULL) ORDER BY lsn DESC LIMIT 1;"
+);
+print "waiting to replay $endpos\n";
+
+# Insert some rows after $endpos, which we won't read.
+$node_primary->safe_psql('testdb',
+ qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(5,50) s;]
+);
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+my $stdout_recv = $node_standby->pg_recvlogical_upto(
+ 'testdb', 'activeslot', $endpos, 180,
+ 'include-xids' => '0',
+ 'skip-empty-xacts' => '1');
+chomp($stdout_recv);
+is($stdout_recv, $expected,
+ 'got same expected output from pg_recvlogical decoding session');
+
+$node_standby->poll_query_until('testdb',
+ "SELECT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = 'activeslot' AND active_pid IS NULL)"
+) or die "slot never became inactive";
+
+$stdout_recv = $node_standby->pg_recvlogical_upto(
+ 'testdb', 'activeslot', $endpos, 180,
+ 'include-xids' => '0',
+ 'skip-empty-xacts' => '1');
+chomp($stdout_recv);
+is($stdout_recv, '', 'pg_recvlogical acknowledged changes');
+
+$node_primary->safe_psql('postgres', 'CREATE DATABASE otherdb');
+
+is( $node_primary->psql(
+ 'otherdb',
+ "SELECT lsn FROM pg_logical_slot_peek_changes('activeslot', NULL, NULL) ORDER BY lsn DESC LIMIT 1;"
+ ),
+ 3,
+ 'replaying logical slot from another database fails');
+
+# drop the logical slots
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('inactiveslot')]);
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('activeslot')]);
+
+##################################################
+# Recovery conflict: Invalidate conflicting slots, including in-use slots
+# Scenario 1: hot_standby_feedback off and vacuum FULL
+##################################################
+
+create_logical_slots();
+
+# One way to reproduce recovery conflict is to run VACUUM FULL with
+# hot_standby_feedback turned off on the standby.
+$node_standby->append_conf('postgresql.conf',q[
+hot_standby_feedback = off
+]);
+$node_standby->restart;
+# ensure walreceiver feedback off by waiting for expected xmin and
+# catalog_xmin on primary. Both should be NULL since hs_feedback is off
+wait_for_xmins($node_primary, $primary_slotname,
+ "xmin IS NULL AND catalog_xmin IS NULL");
+
+$handle = make_slot_active(1);
+
+# This should trigger the conflict
+$node_primary->safe_psql('testdb', 'VACUUM FULL');
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+# message should be issued
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"inactiveslot\" because it conflicts with recovery"),
+ 'inactiveslot slot invalidation is logged with vacuum FULL');
+
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"activeslot\" because it conflicts with recovery"),
+ 'activeslot slot invalidation is logged with vacuum FULL');
+
+# Verify that pg_stat_database_conflicts.confl_logicalslot has been updated
+ok( $node_standby->poll_query_until(
+ 'postgres',
+ "select (confl_logicalslot = 2) from pg_stat_database_conflicts where datname = 'testdb'", 't'),
+ 'confl_logicalslot updated') or die "Timed out waiting confl_logicalslot to be updated";
+
+$handle = make_slot_active(0);
+usleep(100_000);
+
+# We are not able to read from the slot as it has been invalidated
+ok( find_in_log(
+ $node_standby,
+ "cannot read from logical replication slot \"activeslot\""),
+ 'cannot read from logical replication slot');
+
+# Turn hot_standby_feedback back on
+$node_standby->append_conf('postgresql.conf',q[
+hot_standby_feedback = on
+]);
+$node_standby->restart;
+
+# ensure walreceiver feedback sent by waiting for expected xmin and
+# catalog_xmin on primary. With hot_standby_feedback on, xmin should advance,
+# but catalog_xmin should still remain NULL since there is no logical slot.
+wait_for_xmins($node_primary, $primary_slotname,
+ "xmin IS NOT NULL AND catalog_xmin IS NULL");
+
+##################################################
+# Recovery conflict: Invalidate conflicting slots, including in-use slots
+# Scenario 2: conflict due to row removal with hot_standby_feedback off.
+##################################################
+
+# get the position to search from in the standby logfile
+my $logstart = get_log_size($node_standby);
+
+# drop the logical slots
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('inactiveslot')]);
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('activeslot')]);
+
+create_logical_slots();
+
+# One way to produce recovery conflict is to create/drop a relation and launch a vacuum
+# with hot_standby_feedback turned off on the standby.
+$node_standby->append_conf('postgresql.conf',q[
+hot_standby_feedback = off
+]);
+$node_standby->restart;
+# ensure walreceiver feedback off by waiting for expected xmin and
+# catalog_xmin on primary. Both should be NULL since hs_feedback is off
+wait_for_xmins($node_primary, $primary_slotname,
+ "xmin IS NULL AND catalog_xmin IS NULL");
+
+$handle = make_slot_active(1);
+
+# This should trigger the conflict
+$node_primary->safe_psql('testdb', qq[CREATE TABLE conflict_test(x integer, y text);]);
+$node_primary->safe_psql('testdb', qq[DROP TABLE conflict_test;]);
+$node_primary->safe_psql('testdb', 'VACUUM');
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+# message should be issued
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"inactiveslot\" because it conflicts with recovery", $logstart),
+ 'inactiveslot slot invalidation is logged due to row removal');
+
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"activeslot\" because it conflicts with recovery", $logstart),
+ 'activeslot slot invalidation is logged due to row removal');
+
+# Verify that pg_stat_database_conflicts.confl_logicalslot has been updated
+ok( $node_standby->poll_query_until(
+ 'postgres',
+ "select (confl_logicalslot = 2) from pg_stat_database_conflicts where datname = 'testdb'", 't'),
+ 'confl_logicalslot updated') or die "Timed out waiting confl_logicalslot to be updated";
+
+$handle = make_slot_active(0);
+usleep(100_000);
+
+# We are not able to read from the slot as it has been invalidated
+ok( find_in_log(
+ $node_standby,
+ "cannot read from logical replication slot \"activeslot\"", $logstart),
+ 'cannot read from logical replication slot');
+
+# Turn hot_standby_feedback back on
+$node_standby->append_conf('postgresql.conf',q[
+hot_standby_feedback = on
+]);
+$node_standby->restart;
+
+# ensure walreceiver feedback sent by waiting for expected xmin and
+# catalog_xmin on primary. With hot_standby_feedback on, xmin should advance,
+# but catalog_xmin should still remain NULL since there is no logical slot.
+wait_for_xmins($node_primary, $primary_slotname,
+ "xmin IS NOT NULL AND catalog_xmin IS NULL");
+
+##################################################
+# Recovery conflict: Invalidate conflicting slots, including in-use slots
+# Scenario 3: incorrect wal_level on primary.
+##################################################
+
+# get the position to search from in the standby logfile
+$logstart = get_log_size($node_standby);
+
+# drop the logical slots
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('inactiveslot')]);
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('activeslot')]);
+
+create_logical_slots();
+
+$handle = make_slot_active(1);
+
+# Make primary wal_level replica. This will trigger slot conflict.
+$node_primary->append_conf('postgresql.conf',q[
+wal_level = 'replica'
+]);
+$node_primary->restart;
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+# message should be issued
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"inactiveslot\" because it conflicts with recovery", $logstart),
+ 'inactiveslot slot invalidation is logged due to wal_level');
+
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"activeslot\" because it conflicts with recovery", $logstart),
+ 'activeslot slot invalidation is logged due to wal_level');
+
+# Verify that pg_stat_database_conflicts.confl_logicalslot has been updated
+ok( $node_standby->poll_query_until(
+ 'postgres',
+ "select (confl_logicalslot = 2) from pg_stat_database_conflicts where datname = 'testdb'", 't'),
+ 'confl_logicalslot updated') or die "Timed out waiting confl_logicalslot to be updated";
+
+$handle = make_slot_active(0);
+usleep(100_000);
+
+ok( find_in_log(
+ $node_standby,
+ "logical decoding on standby requires wal_level >= logical on master", $logstart),
+ 'cannot start replication because wal_level < logical on master');
+
+# Restore primary wal_level
+$node_primary->append_conf('postgresql.conf',q[
+wal_level = 'logical'
+]);
+$node_primary->restart;
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+$handle = make_slot_active(0);
+usleep(100_000);
+
+# as the slot has been invalidated we should not be able to read
+ok( find_in_log(
+ $node_standby,
+ "cannot read from logical replication slot \"activeslot\"", $logstart),
+ 'cannot read from logical replication slot');
+
+##################################################
+# DROP DATABASE should drops it's slots, including active slots.
+##################################################
+
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('inactiveslot')]);
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('activeslot')]);
+create_logical_slots();
+$handle = make_slot_active(1);
+# Create a slot on a database that would not be dropped. This slot should not
+# get dropped.
+$node_standby->create_logical_slot_on_standby($node_primary, 'otherslot', 'postgres');
+
+# dropdb on the primary to verify slots are dropped on standby
+$node_primary->safe_psql('postgres', q[DROP DATABASE testdb]);
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+is($node_standby->safe_psql('postgres',
+ q[SELECT EXISTS(SELECT 1 FROM pg_database WHERE datname = 'testdb')]), 'f',
+ 'database dropped on standby');
+
+check_slots_dropped($handle);
+
+is($node_standby->slot('otherslot')->{'slot_type'}, 'logical',
+ 'otherslot on standby not dropped');
+
+# Cleanup : manually drop the slot that was not dropped.
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('otherslot')]);
+
+##################################################
+# Test standby promotion and logical decoding behavior
+# after the standby gets promoted.
+##################################################
+
+$node_primary->psql('postgres', q[CREATE DATABASE testdb]);
+$node_primary->safe_psql('testdb', qq[CREATE TABLE decoding_test(x integer, y text);]);
+
+# create the logical slots
+create_logical_slots();
+
+# Insert some rows before the promotion
+$node_primary->safe_psql('testdb',
+ qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(1,4) s;]
+);
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+# promote
+$node_standby->promote;
+
+# insert some rows on promoted standby
+$node_standby->safe_psql('testdb',
+ qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(5,7) s;]
+);
+
+
+$expected = q{BEGIN
+table public.decoding_test: INSERT: x[integer]:1 y[text]:'1'
+table public.decoding_test: INSERT: x[integer]:2 y[text]:'2'
+table public.decoding_test: INSERT: x[integer]:3 y[text]:'3'
+table public.decoding_test: INSERT: x[integer]:4 y[text]:'4'
+COMMIT
+BEGIN
+table public.decoding_test: INSERT: x[integer]:5 y[text]:'5'
+table public.decoding_test: INSERT: x[integer]:6 y[text]:'6'
+table public.decoding_test: INSERT: x[integer]:7 y[text]:'7'
+COMMIT};
+
+# check that we are decoding pre and post promotion inserted rows
+$stdout_sql = $node_standby->safe_psql('testdb',
+ qq[SELECT data FROM pg_logical_slot_peek_changes('activeslot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');]
+);
+
+is($stdout_sql, $expected, 'got expected output from SQL decoding session on promoted standby');
--
2.18.4
From 7b10524f2e81e4a0beb160e6297cfb6d601b6d64 Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 26 Oct 2021 14:34:37 +0000
Subject: [PATCH v25 5/6] Doc changes describing details about logical
decoding.
Author: Andres Freund (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
doc/src/sgml/logicaldecoding.sgml | 21 +++++++++++++++++++++
1 file changed, 21 insertions(+)
100.0% doc/src/sgml/
diff --git a/doc/src/sgml/logicaldecoding.sgml b/doc/src/sgml/logicaldecoding.sgml
index b6353c7a12..98fffc4352 100644
--- a/doc/src/sgml/logicaldecoding.sgml
+++ b/doc/src/sgml/logicaldecoding.sgml
@@ -316,6 +316,27 @@ postgres=# select * from pg_logical_slot_get_changes('regression_slot', NULL, NU
may consume changes from a slot at any given time.
</para>
+ <para>
+ A logical replication slot can also be created on a hot standby. To prevent
+ <command>VACUUM</command> from removing required rows from the system
+ catalogs, <varname>hot_standby_feedback</varname> should be set on the
+ standby. In spite of that, if any required rows get removed, the slot gets
+ invalidated. It's highly recommended to use a physical slot between the primary
+ and the standby. Otherwise, hot_standby_feedback will work, but only while the
+ connection is alive (for example a node restart would break it). Existing
+ logical slots on standby also get invalidated if wal_level on primary is reduced to
+ less than 'logical'.
+ </para>
+
+ <para>
+ For a logical slot to be created, it builds a historic snapshot, for which
+ information of all the currently running transactions is essential. On
+ primary, this information is available, but on standby, this information
+ has to be obtained from primary. So, slot creation may wait for some
+ activity to happen on the primary. If the primary is idle, creating a
+ logical slot on standby may take a noticeable time.
+ </para>
+
<caution>
<para>
Replication slots persist across crashes and know nothing about the state
--
2.18.4
From 95cdd939b709a2a1f1462b6a92c6cc201a3adde7 Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 26 Oct 2021 14:35:23 +0000
Subject: [PATCH v25 6/6] Fixing Walsender corner cases with logical decoding
on standby.
The problem is that WalSndWaitForWal() waits for the *replay* LSN to
increase, but gets woken up by walreceiver when new WAL has been
flushed. Which means that typically walsenders will get woken up at the
same time that the startup process will be - which means that by the
time the logical walsender checks GetXLogReplayRecPtr() it's unlikely
that the startup process already replayed the record and updated
XLogCtl->lastReplayedEndRecPtr.
Fixed by making used of a condition variable on the replay position.
---
src/backend/access/transam/xlog.c | 20 +++++++++++++----
src/backend/replication/walsender.c | 29 ++++++++++++++++++-------
src/backend/utils/activity/wait_event.c | 3 +++
src/include/replication/walsender.h | 12 ++++++++++
src/include/utils/wait_event.h | 1 +
5 files changed, 53 insertions(+), 12 deletions(-)
29.6% src/backend/access/transam/
52.7% src/backend/replication/
4.1% src/backend/utils/activity/
11.7% src/include/replication/
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 1e9f94927c..3d125228fe 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -737,6 +737,7 @@ typedef struct XLogCtlData
} XLogCtlData;
static XLogCtlData *XLogCtl = NULL;
+XLogCtlCvData *XLogCtlCv = NULL;
/* a private copy of XLogCtl->Insert.WALInsertLocks, for convenience */
static WALInsertLockPadded *WALInsertLocks = NULL;
@@ -5182,7 +5183,8 @@ void
XLOGShmemInit(void)
{
bool foundCFile,
- foundXLog;
+ foundXLog,
+ foundXLogCv;
char *allocptr;
int i;
ControlFileData *localControlFile;
@@ -5207,14 +5209,17 @@ XLOGShmemInit(void)
XLogCtl = (XLogCtlData *)
ShmemInitStruct("XLOG Ctl", XLOGShmemSize(), &foundXLog);
+ XLogCtlCv = (XLogCtlCvData *)
+ ShmemInitStruct("XLOG Cv Ctl", sizeof(XLogCtlCvData), &foundXLogCv);
+
localControlFile = ControlFile;
ControlFile = (ControlFileData *)
ShmemInitStruct("Control File", sizeof(ControlFileData), &foundCFile);
- if (foundCFile || foundXLog)
+ if (foundCFile || foundXLog || foundXLogCv)
{
- /* both should be present or neither */
- Assert(foundCFile && foundXLog);
+ /* All should be present or neither */
+ Assert(foundCFile && foundXLog && foundXLogCv);
/* Initialize local copy of WALInsertLocks */
WALInsertLocks = XLogCtl->Insert.WALInsertLocks;
@@ -5224,6 +5229,7 @@ XLOGShmemInit(void)
return;
}
memset(XLogCtl, 0, sizeof(XLogCtlData));
+ memset(XLogCtlCv, 0, sizeof(XLogCtlCvData));
/*
* Already have read control file locally, unless in bootstrap mode. Move
@@ -5285,6 +5291,7 @@ XLOGShmemInit(void)
SpinLockInit(&XLogCtl->ulsn_lck);
InitSharedLatch(&XLogCtl->recoveryWakeupLatch);
ConditionVariableInit(&XLogCtl->recoveryNotPausedCV);
+ ConditionVariableInit(&XLogCtlCv->replayedCV);
}
/*
@@ -7684,6 +7691,11 @@ StartupXLOG(void)
XLogCtl->lastReplayedTLI = ThisTimeLineID;
SpinLockRelease(&XLogCtl->info_lck);
+ /*
+ * wake up walsender(s) used by logical decoding on standby.
+ */
+ ConditionVariableBroadcast(&XLogCtlCv->replayedCV);
+
/*
* If rm_redo called XLogRequestWalReceiverReply, then we wake
* up the receiver so that it notices the updated
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 2ea4d27b63..8460413030 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1493,6 +1493,7 @@ WalSndWaitForWal(XLogRecPtr loc)
{
int wakeEvents;
static XLogRecPtr RecentFlushPtr = InvalidXLogRecPtr;
+ XLogCtlCvData *xlogctlcv = XLogCtlCv;
/*
* Fast path to avoid acquiring the spinlock in case we already know we
@@ -1511,7 +1512,6 @@ WalSndWaitForWal(XLogRecPtr loc)
for (;;)
{
- long sleeptime;
/* Clear any already-pending wakeups */
ResetLatch(MyLatch);
@@ -1595,20 +1595,33 @@ WalSndWaitForWal(XLogRecPtr loc)
WalSndKeepaliveIfNecessary();
/*
- * Sleep until something happens or we time out. Also wait for the
- * socket becoming writable, if there's still pending output.
+ * When not in recovery, sleep until something happens or we time out.
+ * Also wait for the socket becoming writable, if there's still pending output.
* Otherwise we might sit on sendable output data while waiting for
* new WAL to be generated. (But if we have nothing to send, we don't
* want to wake on socket-writable.)
*/
- sleeptime = WalSndComputeSleeptime(GetCurrentTimestamp());
+ if (!RecoveryInProgress())
+ {
+ long sleeptime;
+ sleeptime = WalSndComputeSleeptime(GetCurrentTimestamp());
- wakeEvents = WL_SOCKET_READABLE;
+ wakeEvents = WL_SOCKET_READABLE;
- if (pq_is_send_pending())
- wakeEvents |= WL_SOCKET_WRITEABLE;
+ if (pq_is_send_pending())
+ wakeEvents |= WL_SOCKET_WRITEABLE;
- WalSndWait(wakeEvents, sleeptime, WAIT_EVENT_WAL_SENDER_WAIT_WAL);
+ WalSndWait(wakeEvents, sleeptime * 10, WAIT_EVENT_WAL_SENDER_WAIT_WAL);
+ }
+ else
+ /*
+ * We are in the logical decoding on standby case.
+ * We are waiting for the startup process to replay wal record(s).
+ */
+ {
+ ConditionVariablePrepareToSleep(&xlogctlcv->replayedCV);
+ ConditionVariableSleep(&xlogctlcv->replayedCV, WAIT_EVENT_WAL_SENDER_WAIT_REPLAY);
+ }
}
/* reactivate latch so WalSndLoop knows to continue */
diff --git a/src/backend/utils/activity/wait_event.c b/src/backend/utils/activity/wait_event.c
index 4a5b7502f5..92dc17baf1 100644
--- a/src/backend/utils/activity/wait_event.c
+++ b/src/backend/utils/activity/wait_event.c
@@ -448,6 +448,9 @@ pgstat_get_wait_ipc(WaitEventIPC w)
case WAIT_EVENT_WAL_RECEIVER_WAIT_START:
event_name = "WalReceiverWaitStart";
break;
+ case WAIT_EVENT_WAL_SENDER_WAIT_REPLAY:
+ event_name = "WalReceiverWaitReplay";
+ break;
case WAIT_EVENT_XACT_GROUP_UPDATE:
event_name = "XactGroupUpdate";
break;
diff --git a/src/include/replication/walsender.h b/src/include/replication/walsender.h
index 828106933c..7a2c04c937 100644
--- a/src/include/replication/walsender.h
+++ b/src/include/replication/walsender.h
@@ -13,6 +13,7 @@
#define _WALSENDER_H
#include <signal.h>
+#include "storage/condition_variable.h"
/*
* What to do with a snapshot in create replication slot command.
@@ -48,6 +49,17 @@ extern void WalSndWaitStopping(void);
extern void HandleWalSndInitStopping(void);
extern void WalSndRqstFileReload(void);
+/*
+ * shared-memory state for Condition Variable(s)
+ * between the startup process and the walsender.
+ */
+typedef struct XLogCtlCvData
+{
+ ConditionVariable replayedCV;
+} XLogCtlCvData;
+
+extern XLogCtlCvData *XLogCtlCv;
+
/*
* Remember that we want to wakeup walsenders later
*
diff --git a/src/include/utils/wait_event.h b/src/include/utils/wait_event.h
index c22142365f..b1a27f5e84 100644
--- a/src/include/utils/wait_event.h
+++ b/src/include/utils/wait_event.h
@@ -125,6 +125,7 @@ typedef enum
WAIT_EVENT_SYNC_REP,
WAIT_EVENT_WAL_RECEIVER_EXIT,
WAIT_EVENT_WAL_RECEIVER_WAIT_START,
+ WAIT_EVENT_WAL_SENDER_WAIT_REPLAY,
WAIT_EVENT_XACT_GROUP_UPDATE
} WaitEventIPC;
--
2.18.4
Attachments:
[text/plain] v25-0001-Add-info-in-WAL-records-in-preparation-for-logic.patch (18.0K, ../[email protected]/3-v25-0001-Add-info-in-WAL-records-in-preparation-for-logic.patch)
download | inline diff:
From 748b1104bf2f5acfd9936e486412510d228039cd Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 26 Oct 2021 14:30:22 +0000
Subject: [PATCH v25 1/6] Add info in WAL records in preparation for logical
slot conflict handling.
When a WAL replay on standby indicates that a catalog table tuple is
to be deleted by an xid that is greater than a logical slot's
catalog_xmin, then that means the slot's catalog_xmin conflicts with
the xid, and we need to handle the conflict. While subsequent commits
will do the actual conflict handling, this commit adds a new field
onCatalogTable in such WAL records, that is true for catalog tables,
so as to arrange for conflict handling.
Author: Andres Freund (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
src/backend/access/gist/gist.c | 2 +-
src/backend/access/gist/gistbuild.c | 2 +-
src/backend/access/gist/gistutil.c | 4 ++--
src/backend/access/gist/gistxlog.c | 4 +++-
src/backend/access/hash/hashinsert.c | 1 +
src/backend/access/heap/heapam.c | 4 +++-
src/backend/access/heap/pruneheap.c | 1 +
src/backend/access/heap/visibilitymap.c | 2 +-
src/backend/access/nbtree/nbtpage.c | 12 +++++++++---
src/backend/access/spgist/spgvacuum.c | 8 ++++++++
src/backend/utils/cache/lsyscache.c | 15 +++++++++++++++
src/include/access/gist_private.h | 6 +++---
src/include/access/gistxlog.h | 3 ++-
src/include/access/hash_xlog.h | 1 +
src/include/access/heapam_xlog.h | 5 ++++-
src/include/access/nbtxlog.h | 2 ++
src/include/access/spgxlog.h | 1 +
src/include/utils/lsyscache.h | 1 +
src/include/utils/rel.h | 9 +++++++++
19 files changed, 68 insertions(+), 15 deletions(-)
17.9% src/backend/access/gist/
13.1% src/backend/access/heap/
14.5% src/backend/access/nbtree/
8.9% src/backend/access/spgist/
7.2% src/backend/utils/cache/
19.6% src/include/access/
16.6% src/include/utils/
diff --git a/src/backend/access/gist/gist.c b/src/backend/access/gist/gist.c
index 0683f42c25..b6e6340c3c 100644
--- a/src/backend/access/gist/gist.c
+++ b/src/backend/access/gist/gist.c
@@ -348,7 +348,7 @@ gistplacetopage(Relation rel, Size freespace, GISTSTATE *giststate,
for (; ptr; ptr = ptr->next)
{
/* Allocate new page */
- ptr->buffer = gistNewBuffer(rel);
+ ptr->buffer = gistNewBuffer(heapRel, rel);
GISTInitBuffer(ptr->buffer, (is_leaf) ? F_LEAF : 0);
ptr->page = BufferGetPage(ptr->buffer);
ptr->block.blkno = BufferGetBlockNumber(ptr->buffer);
diff --git a/src/backend/access/gist/gistbuild.c b/src/backend/access/gist/gistbuild.c
index baad28c09f..6d948548c3 100644
--- a/src/backend/access/gist/gistbuild.c
+++ b/src/backend/access/gist/gistbuild.c
@@ -290,7 +290,7 @@ gistbuild(Relation heap, Relation index, IndexInfo *indexInfo)
Page page;
/* initialize the root page */
- buffer = gistNewBuffer(index);
+ buffer = gistNewBuffer(heap, index);
Assert(BufferGetBlockNumber(buffer) == GIST_ROOT_BLKNO);
page = BufferGetPage(buffer);
diff --git a/src/backend/access/gist/gistutil.c b/src/backend/access/gist/gistutil.c
index 43ba03b6eb..1d1e21112c 100644
--- a/src/backend/access/gist/gistutil.c
+++ b/src/backend/access/gist/gistutil.c
@@ -820,7 +820,7 @@ gistcheckpage(Relation rel, Buffer buf)
* Caller is responsible for initializing the page by calling GISTInitBuffer
*/
Buffer
-gistNewBuffer(Relation r)
+gistNewBuffer(Relation heapRel, Relation r)
{
Buffer buffer;
bool needLock;
@@ -864,7 +864,7 @@ gistNewBuffer(Relation r)
* page's deleteXid.
*/
if (XLogStandbyInfoActive() && RelationNeedsWAL(r))
- gistXLogPageReuse(r, blkno, GistPageGetDeleteXid(page));
+ gistXLogPageReuse(heapRel, r, blkno, GistPageGetDeleteXid(page));
return buffer;
}
diff --git a/src/backend/access/gist/gistxlog.c b/src/backend/access/gist/gistxlog.c
index 6464cb9281..46aee6f2a9 100644
--- a/src/backend/access/gist/gistxlog.c
+++ b/src/backend/access/gist/gistxlog.c
@@ -596,7 +596,8 @@ gistXLogAssignLSN(void)
* Write XLOG record about reuse of a deleted page.
*/
void
-gistXLogPageReuse(Relation rel, BlockNumber blkno, FullTransactionId latestRemovedXid)
+gistXLogPageReuse(Relation heapRel, Relation rel,
+ BlockNumber blkno, FullTransactionId latestRemovedXid)
{
gistxlogPageReuse xlrec_reuse;
@@ -607,6 +608,7 @@ gistXLogPageReuse(Relation rel, BlockNumber blkno, FullTransactionId latestRemov
*/
/* XLOG stuff */
+ xlrec_reuse.onCatalogTable = RelationIsAccessibleInLogicalDecoding(heapRel);
xlrec_reuse.node = rel->rd_node;
xlrec_reuse.block = blkno;
xlrec_reuse.latestRemovedFullXid = latestRemovedXid;
diff --git a/src/backend/access/hash/hashinsert.c b/src/backend/access/hash/hashinsert.c
index fe9f0df20b..c2b67a25e1 100644
--- a/src/backend/access/hash/hashinsert.c
+++ b/src/backend/access/hash/hashinsert.c
@@ -398,6 +398,7 @@ _hash_vacuum_one_page(Relation rel, Relation hrel, Buffer metabuf, Buffer buf)
xl_hash_vacuum_one_page xlrec;
XLogRecPtr recptr;
+ xlrec.onCatalogTable = RelationIsAccessibleInLogicalDecoding(hrel);
xlrec.latestRemovedXid = latestRemovedXid;
xlrec.ntuples = ndeletable;
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 2da2be1696..3c45c583cd 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -7945,6 +7945,7 @@ log_heap_freeze(Relation reln, Buffer buffer, TransactionId cutoff_xid,
/* nor when there are no tuples to freeze */
Assert(ntuples > 0);
+ xlrec.onCatalogTable = RelationIsAccessibleInLogicalDecoding(reln);
xlrec.cutoff_xid = cutoff_xid;
xlrec.ntuples = ntuples;
@@ -7975,7 +7976,7 @@ log_heap_freeze(Relation reln, Buffer buffer, TransactionId cutoff_xid,
* heap_buffer, if necessary.
*/
XLogRecPtr
-log_heap_visible(RelFileNode rnode, Buffer heap_buffer, Buffer vm_buffer,
+log_heap_visible(Relation rel, Buffer heap_buffer, Buffer vm_buffer,
TransactionId cutoff_xid, uint8 vmflags)
{
xl_heap_visible xlrec;
@@ -7985,6 +7986,7 @@ log_heap_visible(RelFileNode rnode, Buffer heap_buffer, Buffer vm_buffer,
Assert(BufferIsValid(heap_buffer));
Assert(BufferIsValid(vm_buffer));
+ xlrec.onCatalogTable = RelationIsAccessibleInLogicalDecoding(rel);
xlrec.cutoff_xid = cutoff_xid;
xlrec.flags = vmflags;
XLogBeginInsert();
diff --git a/src/backend/access/heap/pruneheap.c b/src/backend/access/heap/pruneheap.c
index db6912e9fa..b848d0e5a3 100644
--- a/src/backend/access/heap/pruneheap.c
+++ b/src/backend/access/heap/pruneheap.c
@@ -323,6 +323,7 @@ heap_page_prune(Relation relation, Buffer buffer,
xlrec.latestRemovedXid = prstate.latestRemovedXid;
xlrec.nredirected = prstate.nredirected;
xlrec.ndead = prstate.ndead;
+ xlrec.onCatalogTable = RelationIsAccessibleInLogicalDecoding(relation);
XLogBeginInsert();
XLogRegisterData((char *) &xlrec, SizeOfHeapPrune);
diff --git a/src/backend/access/heap/visibilitymap.c b/src/backend/access/heap/visibilitymap.c
index 114fbbdd30..2597636bf0 100644
--- a/src/backend/access/heap/visibilitymap.c
+++ b/src/backend/access/heap/visibilitymap.c
@@ -282,7 +282,7 @@ visibilitymap_set(Relation rel, BlockNumber heapBlk, Buffer heapBuf,
if (XLogRecPtrIsInvalid(recptr))
{
Assert(!InRecovery);
- recptr = log_heap_visible(rel->rd_node, heapBuf, vmBuf,
+ recptr = log_heap_visible(rel, heapBuf, vmBuf,
cutoff_xid, flags);
/*
diff --git a/src/backend/access/nbtree/nbtpage.c b/src/backend/access/nbtree/nbtpage.c
index 5bc7c3616a..3fd59b2a3b 100644
--- a/src/backend/access/nbtree/nbtpage.c
+++ b/src/backend/access/nbtree/nbtpage.c
@@ -36,6 +36,7 @@
#include "utils/memdebug.h"
#include "utils/memutils.h"
#include "utils/snapmgr.h"
+#include "utils/lsyscache.h"
static BTMetaPageData *_bt_getmeta(Relation rel, Buffer metabuf);
static void _bt_log_reuse_page(Relation rel, BlockNumber blkno,
@@ -43,7 +44,8 @@ static void _bt_log_reuse_page(Relation rel, BlockNumber blkno,
static void _bt_delitems_delete(Relation rel, Buffer buf,
TransactionId latestRemovedXid,
OffsetNumber *deletable, int ndeletable,
- BTVacuumPosting *updatable, int nupdatable);
+ BTVacuumPosting *updatable, int nupdatable,
+ Relation heapRel);
static char *_bt_delitems_update(BTVacuumPosting *updatable, int nupdatable,
OffsetNumber *updatedoffsets,
Size *updatedbuflen, bool needswal);
@@ -836,6 +838,7 @@ _bt_log_reuse_page(Relation rel, BlockNumber blkno, FullTransactionId safexid)
*/
/* XLOG stuff */
+ xlrec_reuse.onCatalogTable = get_rel_logical_catalog(rel->rd_index->indrelid);
xlrec_reuse.node = rel->rd_node;
xlrec_reuse.block = blkno;
xlrec_reuse.latestRemovedFullXid = safexid;
@@ -1296,7 +1299,8 @@ _bt_delitems_vacuum(Relation rel, Buffer buf,
static void
_bt_delitems_delete(Relation rel, Buffer buf, TransactionId latestRemovedXid,
OffsetNumber *deletable, int ndeletable,
- BTVacuumPosting *updatable, int nupdatable)
+ BTVacuumPosting *updatable, int nupdatable,
+ Relation heapRel)
{
Page page = BufferGetPage(buf);
BTPageOpaque opaque;
@@ -1358,6 +1362,8 @@ _bt_delitems_delete(Relation rel, Buffer buf, TransactionId latestRemovedXid,
XLogRecPtr recptr;
xl_btree_delete xlrec_delete;
+ xlrec_delete.onCatalogTable =
+ RelationIsAccessibleInLogicalDecoding(heapRel);
xlrec_delete.latestRemovedXid = latestRemovedXid;
xlrec_delete.ndeleted = ndeletable;
xlrec_delete.nupdated = nupdatable;
@@ -1685,7 +1691,7 @@ _bt_delitems_delete_check(Relation rel, Buffer buf, Relation heapRel,
/* Physically delete tuples (or TIDs) using deletable (or updatable) */
_bt_delitems_delete(rel, buf, latestRemovedXid, deletable, ndeletable,
- updatable, nupdatable);
+ updatable, nupdatable, heapRel);
/* be tidy */
for (int i = 0; i < nupdatable; i++)
diff --git a/src/backend/access/spgist/spgvacuum.c b/src/backend/access/spgist/spgvacuum.c
index 76fb0374c4..3186885d14 100644
--- a/src/backend/access/spgist/spgvacuum.c
+++ b/src/backend/access/spgist/spgvacuum.c
@@ -27,6 +27,7 @@
#include "storage/indexfsm.h"
#include "storage/lmgr.h"
#include "utils/snapmgr.h"
+#include "utils/lsyscache.h"
/* Entry in pending-list of TIDs we need to revisit */
@@ -503,6 +504,13 @@ vacuumRedirectAndPlaceholder(Relation index, Buffer buffer)
spgxlogVacuumRedirect xlrec;
GlobalVisState *vistest;
+ /*
+ * There is no chance of endless recursion even when we are doing catalog
+ * acceses here; because, spgist is never used for catalogs. Check
+ * comments in RelationIsAccessibleInLogicalDecoding().
+ */
+ xlrec.onCatalogTable = get_rel_logical_catalog(index->rd_index->indrelid);
+
xlrec.nToPlaceholder = 0;
xlrec.newestRedirectXid = InvalidTransactionId;
diff --git a/src/backend/utils/cache/lsyscache.c b/src/backend/utils/cache/lsyscache.c
index 4ebaa552a2..2fda238870 100644
--- a/src/backend/utils/cache/lsyscache.c
+++ b/src/backend/utils/cache/lsyscache.c
@@ -18,6 +18,7 @@
#include "access/hash.h"
#include "access/htup_details.h"
#include "access/nbtree.h"
+#include "access/table.h"
#include "bootstrap/bootstrap.h"
#include "catalog/namespace.h"
#include "catalog/pg_am.h"
@@ -2062,6 +2063,20 @@ get_rel_persistence(Oid relid)
return result;
}
+bool
+get_rel_logical_catalog(Oid relid)
+{
+ bool res;
+ Relation rel;
+
+ /* assume previously locked */
+ rel = table_open(relid, NoLock);
+ res = RelationIsAccessibleInLogicalDecoding(rel);
+ table_close(rel, NoLock);
+
+ return res;
+}
+
/* ---------- TRANSFORM CACHE ---------- */
diff --git a/src/include/access/gist_private.h b/src/include/access/gist_private.h
index 553d364e2d..a0f4015556 100644
--- a/src/include/access/gist_private.h
+++ b/src/include/access/gist_private.h
@@ -440,8 +440,8 @@ extern XLogRecPtr gistXLogPageDelete(Buffer buffer,
FullTransactionId xid, Buffer parentBuffer,
OffsetNumber downlinkOffset);
-extern void gistXLogPageReuse(Relation rel, BlockNumber blkno,
- FullTransactionId latestRemovedXid);
+extern void gistXLogPageReuse(Relation heapRel, Relation rel,
+ BlockNumber blkno, FullTransactionId latestRemovedXid);
extern XLogRecPtr gistXLogUpdate(Buffer buffer,
OffsetNumber *todelete, int ntodelete,
@@ -485,7 +485,7 @@ extern bool gistproperty(Oid index_oid, int attno,
extern bool gistfitpage(IndexTuple *itvec, int len);
extern bool gistnospace(Page page, IndexTuple *itvec, int len, OffsetNumber todelete, Size freespace);
extern void gistcheckpage(Relation rel, Buffer buf);
-extern Buffer gistNewBuffer(Relation r);
+extern Buffer gistNewBuffer(Relation heapRel, Relation r);
extern bool gistPageRecyclable(Page page);
extern void gistfillbuffer(Page page, IndexTuple *itup, int len,
OffsetNumber off);
diff --git a/src/include/access/gistxlog.h b/src/include/access/gistxlog.h
index fd5144f258..73999ddc70 100644
--- a/src/include/access/gistxlog.h
+++ b/src/include/access/gistxlog.h
@@ -49,9 +49,9 @@ typedef struct gistxlogPageUpdate
*/
typedef struct gistxlogDelete
{
+ bool onCatalogTable;
TransactionId latestRemovedXid;
uint16 ntodelete; /* number of deleted offsets */
-
/*
* In payload of blk 0 : todelete OffsetNumbers
*/
@@ -97,6 +97,7 @@ typedef struct gistxlogPageDelete
*/
typedef struct gistxlogPageReuse
{
+ bool onCatalogTable;
RelFileNode node;
BlockNumber block;
FullTransactionId latestRemovedFullXid;
diff --git a/src/include/access/hash_xlog.h b/src/include/access/hash_xlog.h
index 4353a32dbb..94c3292c1e 100644
--- a/src/include/access/hash_xlog.h
+++ b/src/include/access/hash_xlog.h
@@ -250,6 +250,7 @@ typedef struct xl_hash_init_bitmap_page
*/
typedef struct xl_hash_vacuum_one_page
{
+ bool onCatalogTable;
TransactionId latestRemovedXid;
int ntuples;
diff --git a/src/include/access/heapam_xlog.h b/src/include/access/heapam_xlog.h
index 27db48184e..eba48b0aee 100644
--- a/src/include/access/heapam_xlog.h
+++ b/src/include/access/heapam_xlog.h
@@ -242,6 +242,7 @@ typedef struct xl_heap_update
*/
typedef struct xl_heap_prune
{
+ bool onCatalogTable;
TransactionId latestRemovedXid;
uint16 nredirected;
uint16 ndead;
@@ -338,6 +339,7 @@ typedef struct xl_heap_freeze_tuple
*/
typedef struct xl_heap_freeze_page
{
+ bool onCatalogTable;
TransactionId cutoff_xid;
uint16 ntuples;
} xl_heap_freeze_page;
@@ -352,6 +354,7 @@ typedef struct xl_heap_freeze_page
*/
typedef struct xl_heap_visible
{
+ bool onCatalogTable;
TransactionId cutoff_xid;
uint8 flags;
} xl_heap_visible;
@@ -413,7 +416,7 @@ extern bool heap_prepare_freeze_tuple(HeapTupleHeader tuple,
bool *totally_frozen);
extern void heap_execute_freeze_tuple(HeapTupleHeader tuple,
xl_heap_freeze_tuple *xlrec_tp);
-extern XLogRecPtr log_heap_visible(RelFileNode rnode, Buffer heap_buffer,
+extern XLogRecPtr log_heap_visible(Relation rel, Buffer heap_buffer,
Buffer vm_buffer, TransactionId cutoff_xid, uint8 flags);
#endif /* HEAPAM_XLOG_H */
diff --git a/src/include/access/nbtxlog.h b/src/include/access/nbtxlog.h
index 0f7731856b..b15aa47f1b 100644
--- a/src/include/access/nbtxlog.h
+++ b/src/include/access/nbtxlog.h
@@ -185,6 +185,7 @@ typedef struct xl_btree_dedup
*/
typedef struct xl_btree_reuse_page
{
+ bool onCatalogTable;
RelFileNode node;
BlockNumber block;
FullTransactionId latestRemovedFullXid;
@@ -232,6 +233,7 @@ typedef struct xl_btree_vacuum
typedef struct xl_btree_delete
{
+ bool onCatalogTable;
TransactionId latestRemovedXid;
uint16 ndeleted;
uint16 nupdated;
diff --git a/src/include/access/spgxlog.h b/src/include/access/spgxlog.h
index 69405b5750..06b91f4d04 100644
--- a/src/include/access/spgxlog.h
+++ b/src/include/access/spgxlog.h
@@ -237,6 +237,7 @@ typedef struct spgxlogVacuumRoot
typedef struct spgxlogVacuumRedirect
{
+ bool onCatalogTable;
uint16 nToPlaceholder; /* number of redirects to make placeholders */
OffsetNumber firstPlaceholder; /* first placeholder tuple to remove */
TransactionId newestRedirectXid; /* newest XID of removed redirects */
diff --git a/src/include/utils/lsyscache.h b/src/include/utils/lsyscache.h
index 77871aaefc..e2a5efed30 100644
--- a/src/include/utils/lsyscache.h
+++ b/src/include/utils/lsyscache.h
@@ -139,6 +139,7 @@ extern char get_rel_relkind(Oid relid);
extern bool get_rel_relispartition(Oid relid);
extern Oid get_rel_tablespace(Oid relid);
extern char get_rel_persistence(Oid relid);
+extern bool get_rel_logical_catalog(Oid relid);
extern Oid get_transform_fromsql(Oid typid, Oid langid, List *trftypes);
extern Oid get_transform_tosql(Oid typid, Oid langid, List *trftypes);
extern bool get_typisdefined(Oid typid);
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index b4faa1c123..648aeacd78 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -27,6 +27,7 @@
#include "storage/smgr.h"
#include "utils/relcache.h"
#include "utils/reltrigger.h"
+#include "catalog/catalog.h"
/*
@@ -365,6 +366,9 @@ typedef struct StdRdOptions
* RelationIsUsedAsCatalogTable
* Returns whether the relation should be treated as a catalog table
* from the pov of logical decoding. Note multiple eval of argument!
+ * This definition should not invoke anything that performs catalog
+ * access; otherwise it may cause infinite recursion. Check the comments
+ * in RelationIsAccessibleInLogicalDecoding() for details.
*/
#define RelationIsUsedAsCatalogTable(relation) \
((relation)->rd_options && \
@@ -653,6 +657,11 @@ RelationGetSmgr(Relation rel)
* RelationIsAccessibleInLogicalDecoding
* True if we need to log enough information to have access via
* decoding snapshot.
+ * This definition should not invoke anything that performs catalog
+ * access. Otherwise, e.g. logging a WAL entry for catalog relation may
+ * invoke this function, which will in turn do catalog access, which may
+ * in turn cause another similar WAL entry to be logged, leading to
+ * infinite recursion.
*/
#define RelationIsAccessibleInLogicalDecoding(relation) \
(XLogLogicalInfoActive() && \
--
2.18.4
[text/plain] v25-0002-Handle-logical-slot-conflicts-on-standby.patch (29.3K, ../[email protected]/4-v25-0002-Handle-logical-slot-conflicts-on-standby.patch)
download | inline diff:
From 24b0541619a2a176e502120c3bb1c8e7d3ad2276 Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 26 Oct 2021 14:31:14 +0000
Subject: [PATCH v25 2/6] Handle logical slot conflicts on standby.
During WAL replay on standby, when slot conflict is identified,
invalidate such slots. Also do the same thing if wal_level on master
is reduced to below logical and there are existing logical slots
on standby. Introduce a new ProcSignalReason value for slot
conflict recovery. Arrange for a new pg_stat_get_activity field:
confl_logicalslot.
Author: Andres Freund (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
doc/src/sgml/monitoring.sgml | 10 +
src/backend/access/gist/gistxlog.c | 4 +-
src/backend/access/hash/hash_xlog.c | 3 +-
src/backend/access/heap/heapam.c | 10 +-
src/backend/access/nbtree/nbtxlog.c | 4 +-
src/backend/access/spgist/spgxlog.c | 1 +
src/backend/access/transam/xlog.c | 13 ++
src/backend/catalog/system_views.sql | 3 +-
src/backend/postmaster/pgstat.c | 20 ++
.../replication/logical/logicalfuncs.c | 7 +-
src/backend/replication/slot.c | 209 ++++++++++++++++++
src/backend/replication/walsender.c | 8 +
src/backend/storage/ipc/procarray.c | 4 +
src/backend/storage/ipc/procsignal.c | 3 +
src/backend/storage/ipc/standby.c | 11 +-
src/backend/tcop/postgres.c | 22 ++
src/backend/utils/adt/pgstatfuncs.c | 16 ++
src/include/catalog/pg_proc.dat | 5 +
src/include/pgstat.h | 2 +
src/include/replication/slot.h | 2 +
src/include/storage/procsignal.h | 1 +
src/include/storage/standby.h | 4 +-
src/test/regress/expected/rules.out | 3 +-
23 files changed, 350 insertions(+), 15 deletions(-)
4.1% src/backend/access/heap/
4.7% src/backend/access/transam/
5.1% src/backend/access/
4.3% src/backend/postmaster/
3.5% src/backend/replication/logical/
52.2% src/backend/replication/
5.6% src/backend/storage/ipc/
7.1% src/backend/tcop/
3.6% src/backend/
6.3% src/include/
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 3173ec2566..bd6723bd63 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -4018,6 +4018,16 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
deadlocks
</para></entry>
</row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>confl_logicalslot</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of queries in this database that have been canceled due to
+ logical slots
+ </para></entry>
+ </row>
</tbody>
</tgroup>
</table>
diff --git a/src/backend/access/gist/gistxlog.c b/src/backend/access/gist/gistxlog.c
index 46aee6f2a9..5963e639d8 100644
--- a/src/backend/access/gist/gistxlog.c
+++ b/src/backend/access/gist/gistxlog.c
@@ -195,7 +195,8 @@ gistRedoDeleteRecord(XLogReaderState *record)
XLogRecGetBlockTag(record, 0, &rnode, NULL, NULL);
- ResolveRecoveryConflictWithSnapshot(xldata->latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xldata->latestRemovedXid,
+ xldata->onCatalogTable, rnode);
}
if (XLogReadBufferForRedo(record, 0, &buffer) == BLK_NEEDS_REDO)
@@ -395,6 +396,7 @@ gistRedoPageReuse(XLogReaderState *record)
*/
if (InHotStandby)
ResolveRecoveryConflictWithSnapshotFullXid(xlrec->latestRemovedFullXid,
+ xlrec->onCatalogTable,
xlrec->node);
}
diff --git a/src/backend/access/hash/hash_xlog.c b/src/backend/access/hash/hash_xlog.c
index 27475fcbd6..e5c6124400 100644
--- a/src/backend/access/hash/hash_xlog.c
+++ b/src/backend/access/hash/hash_xlog.c
@@ -1002,7 +1002,8 @@ hash_xlog_vacuum_one_page(XLogReaderState *record)
RelFileNode rnode;
XLogRecGetBlockTag(record, 0, &rnode, NULL, NULL);
- ResolveRecoveryConflictWithSnapshot(xldata->latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xldata->latestRemovedXid,
+ xldata->onCatalogTable, rnode);
}
action = XLogReadBufferForRedoExtended(record, 0, RBM_NORMAL, true, &buffer);
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 3c45c583cd..43624603a0 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -8423,7 +8423,8 @@ heap_xlog_prune(XLogReaderState *record)
* no queries running for which the removed tuples are still visible.
*/
if (InHotStandby)
- ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid,
+ xlrec->onCatalogTable, rnode);
/*
* If we have a full-page image, restore it (using a cleanup lock) and
@@ -8591,7 +8592,9 @@ heap_xlog_visible(XLogReaderState *record)
* rather than killing the transaction outright.
*/
if (InHotStandby)
- ResolveRecoveryConflictWithSnapshot(xlrec->cutoff_xid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xlrec->cutoff_xid,
+ xlrec->onCatalogTable,
+ rnode);
/*
* Read the heap page, if it still exists. If the heap file has dropped or
@@ -8728,7 +8731,8 @@ heap_xlog_freeze_page(XLogReaderState *record)
TransactionIdRetreat(latestRemovedXid);
XLogRecGetBlockTag(record, 0, &rnode, NULL, NULL);
- ResolveRecoveryConflictWithSnapshot(latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(latestRemovedXid,
+ xlrec->onCatalogTable, rnode);
}
if (XLogReadBufferForRedo(record, 0, &buffer) == BLK_NEEDS_REDO)
diff --git a/src/backend/access/nbtree/nbtxlog.c b/src/backend/access/nbtree/nbtxlog.c
index 786c08c0ce..41b7ec8e2d 100644
--- a/src/backend/access/nbtree/nbtxlog.c
+++ b/src/backend/access/nbtree/nbtxlog.c
@@ -668,7 +668,8 @@ btree_xlog_delete(XLogReaderState *record)
XLogRecGetBlockTag(record, 0, &rnode, NULL, NULL);
- ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid,
+ xlrec->onCatalogTable, rnode);
}
/*
@@ -1006,6 +1007,7 @@ btree_xlog_reuse_page(XLogReaderState *record)
if (InHotStandby)
ResolveRecoveryConflictWithSnapshotFullXid(xlrec->latestRemovedFullXid,
+ xlrec->onCatalogTable,
xlrec->node);
}
diff --git a/src/backend/access/spgist/spgxlog.c b/src/backend/access/spgist/spgxlog.c
index 3dfd2aa317..add4da4e74 100644
--- a/src/backend/access/spgist/spgxlog.c
+++ b/src/backend/access/spgist/spgxlog.c
@@ -881,6 +881,7 @@ spgRedoVacuumRedirect(XLogReaderState *record)
XLogRecGetBlockTag(record, 0, &node, NULL, NULL);
ResolveRecoveryConflictWithSnapshot(xldata->newestRedirectXid,
+ xldata->onCatalogTable,
node);
}
}
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index f547efd294..0d2590ebeb 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -10577,6 +10577,19 @@ xlog_redo(XLogReaderState *record)
/* Update our copy of the parameters in pg_control */
memcpy(&xlrec, XLogRecGetData(record), sizeof(xl_parameter_change));
+ /*
+ * Invalidate logical slots if we are in hot standby and the primary does not
+ * have a WAL level sufficient for logical decoding. No need to search
+ * for potentially conflicting logically slots if standby is running
+ * with wal_level lower than logical, because in that case, we would
+ * have either disallowed creation of logical slots or invalidated existing
+ * ones.
+ */
+ if (InRecovery && InHotStandby &&
+ xlrec.wal_level < WAL_LEVEL_LOGICAL &&
+ wal_level >= WAL_LEVEL_LOGICAL)
+ InvalidateConflictingLogicalReplicationSlots(InvalidOid,InvalidTransactionId);
+
LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
ControlFile->MaxConnections = xlrec.MaxConnections;
ControlFile->max_worker_processes = xlrec.max_worker_processes;
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 55f6e3711d..cd1fc88d17 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1020,7 +1020,8 @@ CREATE VIEW pg_stat_database_conflicts AS
pg_stat_get_db_conflict_lock(D.oid) AS confl_lock,
pg_stat_get_db_conflict_snapshot(D.oid) AS confl_snapshot,
pg_stat_get_db_conflict_bufferpin(D.oid) AS confl_bufferpin,
- pg_stat_get_db_conflict_startup_deadlock(D.oid) AS confl_deadlock
+ pg_stat_get_db_conflict_startup_deadlock(D.oid) AS confl_deadlock,
+ pg_stat_get_db_conflict_logicalslot(D.oid) AS confl_logicalslot
FROM pg_database D;
CREATE VIEW pg_stat_user_functions AS
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index b7d0fbaefd..4fad4a9b1c 100644
--- a/src/backend/postmaster/pgstat.c
+++ b/src/backend/postmaster/pgstat.c
@@ -1869,6 +1869,22 @@ pgstat_report_replslot_drop(const char *slotname)
pgstat_send(&msg, sizeof(PgStat_MsgReplSlot));
}
+/* ----------
+ * pgstat_report_replslot_conflict()
+ * Tell the collector about a logical slot being conflicting
+ * with recovery.
+ * ----------
+ */
+void
+pgstat_report_replslot_conflict(Oid dboid)
+{
+ PgStat_MsgRecoveryConflict msg;
+ pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_RECOVERYCONFLICT);
+ msg.m_databaseid = dboid;
+ msg.m_reason = PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT;
+ pgstat_send(&msg, sizeof(msg));
+}
+
/* ----------
* pgstat_ping() -
*
@@ -3633,6 +3649,7 @@ reset_dbentry_counters(PgStat_StatDBEntry *dbentry)
dbentry->n_conflict_tablespace = 0;
dbentry->n_conflict_lock = 0;
dbentry->n_conflict_snapshot = 0;
+ dbentry->n_conflict_logicalslot = 0;
dbentry->n_conflict_bufferpin = 0;
dbentry->n_conflict_startup_deadlock = 0;
dbentry->n_temp_files = 0;
@@ -5580,6 +5597,9 @@ pgstat_recv_recoveryconflict(PgStat_MsgRecoveryConflict *msg, int len)
case PROCSIG_RECOVERY_CONFLICT_SNAPSHOT:
dbentry->n_conflict_snapshot++;
break;
+ case PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT:
+ dbentry->n_conflict_logicalslot++;
+ break;
case PROCSIG_RECOVERY_CONFLICT_BUFFERPIN:
dbentry->n_conflict_bufferpin++;
break;
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index e59939aad1..805d4c5a5b 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -232,11 +232,14 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
/*
* After the sanity checks in CreateDecodingContext, make sure the
- * restart_lsn is valid. Avoid "cannot get changes" wording in this
+ * restart_lsn is valid or both xmin and catalog_xmin are valid.
+ * Avoid "cannot get changes" wording in this
* errmsg because that'd be confusingly ambiguous about no changes
* being available.
*/
- if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn))
+ if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn)
+ || (!TransactionIdIsValid(MyReplicationSlot->data.xmin)
+ && !TransactionIdIsValid(MyReplicationSlot->data.catalog_xmin)))
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("can no longer get changes from replication slot \"%s\"",
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 1c6c0c7ce2..188716957f 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -1350,6 +1350,215 @@ restart:
return invalidated;
}
+/*
+ * Helper for InvalidateConflictingLogicalReplicationSlot -- acquires the given slot
+ * and mark it invalid, if necessary and possible.
+ *
+ * Returns whether ReplicationSlotControlLock was released in the interim (and
+ * in that case we're not holding the lock at return, otherwise we are).
+ *
+ * This is inherently racy, because we release the LWLock
+ * for syscalls, so caller must restart if we return true.
+ */
+static bool
+InvalidatePossiblyConflictingLogicalReplicationSlot(ReplicationSlot *s, TransactionId xid)
+{
+ int last_signaled_pid = 0;
+ bool released_lock = false;
+
+ for (;;)
+ {
+ TransactionId slot_xmin;
+ TransactionId slot_catalog_xmin;
+ NameData slotname;
+ int active_pid = 0;
+
+ Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
+
+ if (!s->in_use)
+ {
+ if (released_lock)
+ LWLockRelease(ReplicationSlotControlLock);
+ break;
+ }
+
+ /*
+ * Check if the slot needs to be invalidated. If it needs to be
+ * invalidated, and is not currently acquired, acquire it and mark it
+ * as having been invalidated. We do this with the spinlock held to
+ * avoid race conditions -- for example the xmin(s) could move forward
+ * , or the slot could be dropped.
+ */
+ SpinLockAcquire(&s->mutex);
+
+ slot_xmin = s->data.xmin;
+ slot_catalog_xmin = s->data.catalog_xmin;
+
+ /*
+ * If the slot is already invalid or is not conflicting, we don't need to
+ * do anything.
+ */
+
+ /* slot has been invalidated */
+ if ((!TransactionIdIsValid(slot_xmin) && !TransactionIdIsValid(slot_catalog_xmin))
+ ||
+ /*
+ * we are not forcing for invalidation because the xid is valid
+ * and this is a non conflicting slot
+ */
+ (TransactionIdIsValid(xid) && !(
+ (TransactionIdIsValid(slot_xmin) && TransactionIdPrecedesOrEquals(slot_xmin, xid))
+ ||
+ (TransactionIdIsValid(slot_catalog_xmin) && TransactionIdPrecedesOrEquals(slot_catalog_xmin, xid))
+ ))
+ )
+ {
+ SpinLockRelease(&s->mutex);
+ if (released_lock)
+ LWLockRelease(ReplicationSlotControlLock);
+ break;
+ }
+
+ slotname = s->data.name;
+ active_pid = s->active_pid;
+
+ /*
+ * If the slot can be acquired, do so and mark it invalidated
+ * immediately. Otherwise we'll signal the owning process, below, and
+ * retry.
+ */
+ if (active_pid == 0)
+ {
+ MyReplicationSlot = s;
+ s->active_pid = MyProcPid;
+ s->data.xmin = InvalidTransactionId;
+ s->data.catalog_xmin = InvalidTransactionId;
+ }
+
+ SpinLockRelease(&s->mutex);
+
+ if (active_pid != 0)
+ {
+ /*
+ * Prepare the sleep on the slot's condition variable before
+ * releasing the lock, to close a possible race condition if the
+ * slot is released before the sleep below.
+ */
+
+ ConditionVariablePrepareToSleep(&s->active_cv);
+
+ LWLockRelease(ReplicationSlotControlLock);
+ released_lock = true;
+
+ /*
+ * Signal to terminate the process that owns the slot, if we
+ * haven't already signalled it. (Avoidance of repeated
+ * signalling is the only reason for there to be a loop in this
+ * routine; otherwise we could rely on caller's restart loop.)
+ *
+ * There is the race condition that other process may own the slot
+ * after its current owner process is terminated and before this
+ * process owns it. To handle that, we signal only if the PID of
+ * the owning process has changed from the previous time. (This
+ * logic assumes that the same PID is not reused very quickly.)
+ */
+ if (last_signaled_pid != active_pid)
+ {
+ ereport(LOG,
+ (errmsg("terminating process %d because replication slot \"%s\" conflicts with recovery",
+ active_pid, NameStr(slotname))));
+
+ (void) kill(active_pid, SIGTERM);
+ last_signaled_pid = active_pid;
+ }
+
+ /* Wait until the slot is released. */
+ ConditionVariableSleep(&s->active_cv,
+ WAIT_EVENT_REPLICATION_SLOT_DROP);
+
+ /*
+ * Re-acquire lock and start over; we expect to invalidate the
+ * slot next time (unless another process acquires the slot in the
+ * meantime).
+ */
+ LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+ continue;
+ }
+ else
+ {
+ /*
+ * We hold the slot now and have already invalidated it; flush it
+ * to ensure that state persists.
+ *
+ * Don't want to hold ReplicationSlotControlLock across file
+ * system operations, so release it now but be sure to tell caller
+ * to restart from scratch.
+ */
+ LWLockRelease(ReplicationSlotControlLock);
+ released_lock = true;
+
+ /* Make sure the invalidated state persists across server restart */
+ ReplicationSlotMarkDirty();
+ ReplicationSlotSave();
+ ReplicationSlotRelease();
+ pgstat_report_replslot_conflict(s->data.database);
+
+ ereport(LOG,
+ (errmsg("invalidating slot \"%s\" because it conflicts with recovery", NameStr(slotname))));
+
+ /* done with this slot for now */
+ break;
+ }
+ }
+
+ Assert(!released_lock == LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
+
+ return released_lock;
+}
+
+/*
+ * Resolve recovery conflicts with logical slots.
+ *
+ * When xid is valid, it means that we are about to remove rows older than xid.
+ * Therefore we need to invalidate slots that depend on seeing those rows.
+ * When xid is invalid, invalidate all logical slots. This is required when the
+ * master wal_level is set back to replica, so existing logical slots need to
+ * be invalidated.
+ */
+void
+InvalidateConflictingLogicalReplicationSlots(Oid dboid, TransactionId xid)
+{
+
+ Assert(max_replication_slots >= 0);
+
+ if (max_replication_slots == 0)
+ return;
+restart:
+ LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+ for (int i = 0; i < max_replication_slots; i++)
+ {
+ ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
+
+ if (!s->in_use)
+ continue;
+
+ /* We are only dealing with *logical* slot conflicts. */
+ if (!SlotIsLogical(s))
+ continue;
+
+ /* not our database and we don't want all the database, skip */
+ if (s->data.database != dboid && TransactionIdIsValid(xid))
+ continue;
+
+ if (InvalidatePossiblyConflictingLogicalReplicationSlot(s, xid))
+ {
+ /* if the lock was released, we need to restart from scratch */
+ goto restart;
+ }
+ }
+ LWLockRelease(ReplicationSlotControlLock);
+}
+
/*
* Flush all replication slots to disk.
*
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index d9ab6d6de2..63c0dd2f62 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1246,6 +1246,14 @@ StartLogicalReplication(StartReplicationCmd *cmd)
ReplicationSlotAcquire(cmd->slotname, true);
+ if (!TransactionIdIsValid(MyReplicationSlot->data.xmin)
+ && !TransactionIdIsValid(MyReplicationSlot->data.catalog_xmin))
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot read from logical replication slot \"%s\"",
+ cmd->slotname),
+ errdetail("This slot has been invalidated because it was conflicting with recovery.")));
+
if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn))
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index bd3c7a47fe..23ffc39a0a 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -3443,6 +3443,10 @@ SignalVirtualTransaction(VirtualTransactionId vxid, ProcSignalReason sigmode,
GET_VXID_FROM_PGPROC(procvxid, *proc);
+ /*
+ * Note: vxid.localTransactionId can be invalid, which means the
+ * request is to signal the pid that is not running a transaction.
+ */
if (procvxid.backendId == vxid.backendId &&
procvxid.localTransactionId == vxid.localTransactionId)
{
diff --git a/src/backend/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c
index defb75aa26..315d5a1e33 100644
--- a/src/backend/storage/ipc/procsignal.c
+++ b/src/backend/storage/ipc/procsignal.c
@@ -673,6 +673,9 @@ procsignal_sigusr1_handler(SIGNAL_ARGS)
if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_SNAPSHOT))
RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_SNAPSHOT);
+ if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT))
+ RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT);
+
if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK))
RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK);
diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c
index b17326bc20..111fec1b9f 100644
--- a/src/backend/storage/ipc/standby.c
+++ b/src/backend/storage/ipc/standby.c
@@ -34,6 +34,7 @@
#include "utils/ps_status.h"
#include "utils/timeout.h"
#include "utils/timestamp.h"
+#include "replication/slot.h"
/* User-settable GUC parameters */
int vacuum_defer_cleanup_age;
@@ -440,7 +441,8 @@ ResolveRecoveryConflictWithVirtualXIDs(VirtualTransactionId *waitlist,
}
void
-ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid, RelFileNode node)
+ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid,
+ bool onCatalogTable, RelFileNode node)
{
VirtualTransactionId *backends;
@@ -465,6 +467,9 @@ ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid, RelFileNode
PROCSIG_RECOVERY_CONFLICT_SNAPSHOT,
WAIT_EVENT_RECOVERY_CONFLICT_SNAPSHOT,
true);
+
+ if (onCatalogTable)
+ InvalidateConflictingLogicalReplicationSlots(node.dbNode, latestRemovedXid);
}
/*
@@ -473,7 +478,7 @@ ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid, RelFileNode
*/
void
ResolveRecoveryConflictWithSnapshotFullXid(FullTransactionId latestRemovedFullXid,
- RelFileNode node)
+ bool onCatalogTable, RelFileNode node)
{
/*
* ResolveRecoveryConflictWithSnapshot operates on 32-bit TransactionIds,
@@ -491,7 +496,7 @@ ResolveRecoveryConflictWithSnapshotFullXid(FullTransactionId latestRemovedFullXi
TransactionId latestRemovedXid;
latestRemovedXid = XidFromFullTransactionId(latestRemovedFullXid);
- ResolveRecoveryConflictWithSnapshot(latestRemovedXid, node);
+ ResolveRecoveryConflictWithSnapshot(latestRemovedXid, onCatalogTable, node);
}
}
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index 0775abe35d..697b0ea7ad 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -2467,6 +2467,9 @@ errdetail_recovery_conflict(void)
case PROCSIG_RECOVERY_CONFLICT_SNAPSHOT:
errdetail("User query might have needed to see row versions that must be removed.");
break;
+ case PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT:
+ errdetail("User was using the logical slot that must be dropped.");
+ break;
case PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK:
errdetail("User transaction caused buffer deadlock with recovery.");
break;
@@ -3037,6 +3040,25 @@ RecoveryConflictInterrupt(ProcSignalReason reason)
case PROCSIG_RECOVERY_CONFLICT_LOCK:
case PROCSIG_RECOVERY_CONFLICT_TABLESPACE:
case PROCSIG_RECOVERY_CONFLICT_SNAPSHOT:
+ case PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT:
+ /*
+ * For conflicts that require a logical slot to be invalidated, the
+ * requirement is for the signal receiver to release the slot,
+ * so that it could be invalidated by the signal sender. So for
+ * normal backends, the transaction should be aborted, just
+ * like for other recovery conflicts. But if it's walsender on
+ * standby, then it has to be killed so as to release an
+ * acquired logical slot.
+ */
+ if (am_cascading_walsender &&
+ reason == PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT &&
+ MyReplicationSlot && SlotIsLogical(MyReplicationSlot))
+ {
+ RecoveryConflictPending = true;
+ QueryCancelPending = true;
+ InterruptPending = true;
+ break;
+ }
/*
* If we aren't in a transaction any longer then ignore.
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index ff5aedc99c..073b402bac 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -1497,6 +1497,21 @@ pg_stat_get_db_conflict_snapshot(PG_FUNCTION_ARGS)
PG_RETURN_INT64(result);
}
+Datum
+pg_stat_get_db_conflict_logicalslot(PG_FUNCTION_ARGS)
+{
+ Oid dbid = PG_GETARG_OID(0);
+ int64 result;
+ PgStat_StatDBEntry *dbentry;
+
+ if ((dbentry = pgstat_fetch_stat_dbentry(dbid)) == NULL)
+ result = 0;
+ else
+ result = (int64) (dbentry->n_conflict_logicalslot);
+
+ PG_RETURN_INT64(result);
+}
+
Datum
pg_stat_get_db_conflict_bufferpin(PG_FUNCTION_ARGS)
{
@@ -1540,6 +1555,7 @@ pg_stat_get_db_conflict_all(PG_FUNCTION_ARGS)
result = (int64) (dbentry->n_conflict_tablespace +
dbentry->n_conflict_lock +
dbentry->n_conflict_snapshot +
+ dbentry->n_conflict_logicalslot +
dbentry->n_conflict_bufferpin +
dbentry->n_conflict_startup_deadlock);
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index d068d6532e..e40f57a549 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5502,6 +5502,11 @@
proname => 'pg_stat_get_db_conflict_snapshot', provolatile => 's',
proparallel => 'r', prorettype => 'int8', proargtypes => 'oid',
prosrc => 'pg_stat_get_db_conflict_snapshot' },
+{ oid => '4544',
+ descr => 'statistics: recovery conflicts in database caused by logical replication slot',
+ proname => 'pg_stat_get_db_conflict_logicalslot', provolatile => 's',
+ proparallel => 'r', prorettype => 'int8', proargtypes => 'oid',
+ prosrc => 'pg_stat_get_db_conflict_logicalslot' },
{ oid => '3068',
descr => 'statistics: recovery conflicts in database caused by shared buffer pin',
proname => 'pg_stat_get_db_conflict_bufferpin', provolatile => 's',
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index bcd3588ea2..3570e4933a 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -747,6 +747,7 @@ typedef struct PgStat_StatDBEntry
PgStat_Counter n_conflict_tablespace;
PgStat_Counter n_conflict_lock;
PgStat_Counter n_conflict_snapshot;
+ PgStat_Counter n_conflict_logicalslot;
PgStat_Counter n_conflict_bufferpin;
PgStat_Counter n_conflict_startup_deadlock;
PgStat_Counter n_temp_files;
@@ -1038,6 +1039,7 @@ extern void pgstat_report_checksum_failure(void);
extern void pgstat_report_replslot(const PgStat_StatReplSlotEntry *repSlotStat);
extern void pgstat_report_replslot_create(const char *slotname);
extern void pgstat_report_replslot_drop(const char *slotname);
+extern void pgstat_report_replslot_conflict(Oid dbOid);
extern void pgstat_initialize(void);
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 53d773ccff..cca4bb058c 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -214,6 +214,7 @@ extern XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN(void);
extern bool ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive);
extern void ReplicationSlotsDropDBSlots(Oid dboid);
extern bool InvalidateObsoleteReplicationSlots(XLogSegNo oldestSegno);
+extern void InvalidateConflictingLogicalReplicationSlots(Oid dboid, TransactionId xid);
extern ReplicationSlot *SearchNamedReplicationSlot(const char *name, bool need_lock);
extern void ReplicationSlotNameForTablesync(Oid suboid, Oid relid, char *syncslotname, int szslot);
extern void ReplicationSlotDropAtPubNode(WalReceiverConn *wrconn, char *slotname, bool missing_ok);
@@ -223,5 +224,6 @@ extern void CheckPointReplicationSlots(void);
extern void CheckSlotRequirements(void);
extern void CheckSlotPermissions(void);
+extern void ResolveRecoveryConflictWithLogicalSlots(Oid dboid, TransactionId xid, char *reason);
#endif /* SLOT_H */
diff --git a/src/include/storage/procsignal.h b/src/include/storage/procsignal.h
index eec186be2e..f86b070dbc 100644
--- a/src/include/storage/procsignal.h
+++ b/src/include/storage/procsignal.h
@@ -41,6 +41,7 @@ typedef enum
PROCSIG_RECOVERY_CONFLICT_TABLESPACE,
PROCSIG_RECOVERY_CONFLICT_LOCK,
PROCSIG_RECOVERY_CONFLICT_SNAPSHOT,
+ PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT,
PROCSIG_RECOVERY_CONFLICT_BUFFERPIN,
PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK,
diff --git a/src/include/storage/standby.h b/src/include/storage/standby.h
index 38fd85a431..3ba1882216 100644
--- a/src/include/storage/standby.h
+++ b/src/include/storage/standby.h
@@ -30,9 +30,9 @@ extern void InitRecoveryTransactionEnvironment(void);
extern void ShutdownRecoveryTransactionEnvironment(void);
extern void ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid,
- RelFileNode node);
+ bool onCatalogTable, RelFileNode node);
extern void ResolveRecoveryConflictWithSnapshotFullXid(FullTransactionId latestRemovedFullXid,
- RelFileNode node);
+ bool onCatalogTable, RelFileNode node);
extern void ResolveRecoveryConflictWithTablespace(Oid tsid);
extern void ResolveRecoveryConflictWithDatabase(Oid dbid);
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 2fa00a3c29..7cf918634c 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1871,7 +1871,8 @@ pg_stat_database_conflicts| SELECT d.oid AS datid,
pg_stat_get_db_conflict_lock(d.oid) AS confl_lock,
pg_stat_get_db_conflict_snapshot(d.oid) AS confl_snapshot,
pg_stat_get_db_conflict_bufferpin(d.oid) AS confl_bufferpin,
- pg_stat_get_db_conflict_startup_deadlock(d.oid) AS confl_deadlock
+ pg_stat_get_db_conflict_startup_deadlock(d.oid) AS confl_deadlock,
+ pg_stat_get_db_conflict_logicalslot(d.oid) AS confl_logicalslot
FROM pg_database d;
pg_stat_gssapi| SELECT s.pid,
s.gss_auth AS gss_authenticated,
--
2.18.4
[text/plain] v25-0003-Allow-logical-decoding-on-standby.patch (17.5K, ../[email protected]/5-v25-0003-Allow-logical-decoding-on-standby.patch)
download | inline diff:
From 178a6d1c8fecf13164a5b65a8d5677b92dc40aa6 Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 26 Oct 2021 14:32:11 +0000
Subject: [PATCH v25 3/6] Allow logical decoding on standby.
Allow a logical slot to be created on standby. Restrict its usage
or its creation if wal_level on primary is less than logical.
During slot creation, it's restart_lsn is set to the last replayed
LSN. Effectively, a logical slot creation on standby waits for an
xl_running_xact record to arrive from primary. Conflicting slots
would be handled in next commits.
Author: Andres Freund (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
src/backend/access/transam/xlog.c | 33 ++++++++++-
src/backend/access/transam/xlogfuncs.c | 2 +-
src/backend/access/transam/xlogutils.c | 2 +-
src/backend/postmaster/checkpointer.c | 4 +-
src/backend/replication/logical/decode.c | 22 +++++++-
src/backend/replication/logical/logical.c | 37 ++++++------
.../replication/logical/logicalfuncs.c | 2 +-
src/backend/replication/slot.c | 56 ++++++++++---------
src/backend/replication/slotfuncs.c | 2 +-
src/backend/replication/walreceiver.c | 4 +-
src/backend/replication/walreceiverfuncs.c | 2 +-
src/backend/replication/walsender.c | 28 +++++++---
src/include/access/xlog.h | 3 +-
13 files changed, 133 insertions(+), 64 deletions(-)
16.4% src/backend/access/transam/
32.2% src/backend/replication/logical/
46.0% src/backend/replication/
5.2% src/
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 0d2590ebeb..1e9f94927c 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -5123,6 +5123,17 @@ LocalProcessControlFile(bool reset)
ReadControlFile();
}
+/*
+ * Get the wal_level from the control file. For a standby, this value should be
+ * considered as its active wal_level, because it may be different from what
+ * was originally configured on standby.
+ */
+WalLevel
+GetActiveWalLevelOnStandby(void)
+{
+ return ControlFile->wal_level;
+}
+
/*
* Initialization of shared memory for XLOG
*/
@@ -9787,7 +9798,7 @@ CreateRestartPoint(int flags)
* whichever is later.
*/
receivePtr = GetWalRcvFlushRecPtr(NULL, NULL);
- replayPtr = GetXLogReplayRecPtr(&replayTLI);
+ replayPtr = GetXLogReplayRecPtr(&replayTLI, false);
endptr = (receivePtr < replayPtr) ? replayPtr : receivePtr;
KeepLogSeg(endptr, &_logSegNo);
if (InvalidateObsoleteReplicationSlots(_logSegNo))
@@ -11952,7 +11963,7 @@ register_persistent_abort_backup_handler(void)
* Exported to allow WALReceiver to read the pointer directly.
*/
XLogRecPtr
-GetXLogReplayRecPtr(TimeLineID *replayTLI)
+GetXLogReplayRecPtr(TimeLineID *replayTLI, bool avoid_header)
{
XLogRecPtr recptr;
TimeLineID tli;
@@ -11964,6 +11975,24 @@ GetXLogReplayRecPtr(TimeLineID *replayTLI)
if (replayTLI)
*replayTLI = tli;
+
+ /*
+ * Replay pointer may point one past the end of the record. If that
+ * is a XLOG page boundary, it will not be a valid LSN for the
+ * start of a record, so bump it up past the page header.
+ */
+ if (avoid_header && !XRecOffIsValid(recptr))
+ {
+ if (recptr % XLOG_BLCKSZ != 0)
+ elog(ERROR, "invalid replay pointer");
+
+ /* For the first page of a segment file, it's a long header */
+ if (XLogSegmentOffset(recptr, wal_segment_size) == 0)
+ recptr += SizeOfXLogLongPHD;
+ else
+ recptr += SizeOfXLogShortPHD;
+ }
+
return recptr;
}
diff --git a/src/backend/access/transam/xlogfuncs.c b/src/backend/access/transam/xlogfuncs.c
index b98deb72ec..a173f8d6fc 100644
--- a/src/backend/access/transam/xlogfuncs.c
+++ b/src/backend/access/transam/xlogfuncs.c
@@ -417,7 +417,7 @@ pg_last_wal_replay_lsn(PG_FUNCTION_ARGS)
{
XLogRecPtr recptr;
- recptr = GetXLogReplayRecPtr(NULL);
+ recptr = GetXLogReplayRecPtr(NULL, false);
if (recptr == 0)
PG_RETURN_NULL();
diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c
index 88a1bfd939..aaade9382d 100644
--- a/src/backend/access/transam/xlogutils.c
+++ b/src/backend/access/transam/xlogutils.c
@@ -870,7 +870,7 @@ read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
if (!RecoveryInProgress())
read_upto = GetFlushRecPtr();
else
- read_upto = GetXLogReplayRecPtr(&ThisTimeLineID);
+ read_upto = GetXLogReplayRecPtr(&ThisTimeLineID, false);
tli = ThisTimeLineID;
/*
diff --git a/src/backend/postmaster/checkpointer.c b/src/backend/postmaster/checkpointer.c
index be7366379d..ea1bf7d247 100644
--- a/src/backend/postmaster/checkpointer.c
+++ b/src/backend/postmaster/checkpointer.c
@@ -434,7 +434,7 @@ CheckpointerMain(void)
*/
ckpt_active = true;
if (do_restartpoint)
- ckpt_start_recptr = GetXLogReplayRecPtr(NULL);
+ ckpt_start_recptr = GetXLogReplayRecPtr(NULL, false);
else
ckpt_start_recptr = GetInsertRecPtr();
ckpt_start_time = now;
@@ -794,7 +794,7 @@ IsCheckpointOnSchedule(double progress)
* value that was in effect when the WAL was generated).
*/
if (RecoveryInProgress())
- recptr = GetXLogReplayRecPtr(NULL);
+ recptr = GetXLogReplayRecPtr(NULL, false);
else
recptr = GetInsertRecPtr();
elapsed_xlogs = (((double) (recptr - ckpt_start_recptr)) /
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index 2874dc0612..b8be0c83ed 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -215,11 +215,31 @@ DecodeXLogOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
* can restart from there.
*/
break;
+ case XLOG_PARAMETER_CHANGE:
+ {
+ xl_parameter_change *xlrec =
+ (xl_parameter_change *) XLogRecGetData(buf->record);
+
+ /*
+ * If wal_level on primary is reduced to less than logical, then we
+ * want to prevent existing logical slots from being used.
+ * Existing logical slots on standby get invalidated when this WAL
+ * record is replayed; and further, slot creation fails when the
+ * wal level is not sufficient; but all these operations are not
+ * synchronized, so a logical slot may creep in while the wal_level
+ * is being reduced. Hence this extra check.
+ */
+ if (xlrec->wal_level < WAL_LEVEL_LOGICAL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("logical decoding on standby requires "
+ "wal_level >= logical on master")));
+ break;
+ }
case XLOG_NOOP:
case XLOG_NEXTOID:
case XLOG_SWITCH:
case XLOG_BACKUP_END:
- case XLOG_PARAMETER_CHANGE:
case XLOG_RESTORE_POINT:
case XLOG_FPW_CHANGE:
case XLOG_FPI_FOR_HINT:
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index aae0ae5b8a..1e8b2808d5 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -119,23 +119,22 @@ CheckLogicalDecodingRequirements(void)
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("logical decoding requires a database connection")));
- /* ----
- * TODO: We got to change that someday soon...
- *
- * There's basically three things missing to allow this:
- * 1) We need to be able to correctly and quickly identify the timeline a
- * LSN belongs to
- * 2) We need to force hot_standby_feedback to be enabled at all times so
- * the primary cannot remove rows we need.
- * 3) support dropping replication slots referring to a database, in
- * dbase_redo. There can't be any active ones due to HS recovery
- * conflicts, so that should be relatively easy.
- * ----
- */
if (RecoveryInProgress())
- ereport(ERROR,
- (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
- errmsg("logical decoding cannot be used while in recovery")));
+ {
+ /*
+ * This check may have race conditions, but whenever
+ * XLOG_PARAMETER_CHANGE indicates that wal_level has changed, we
+ * verify that there are no existing logical replication slots. And to
+ * avoid races around creating a new slot,
+ * CheckLogicalDecodingRequirements() is called once before creating
+ * the slot, and once when logical decoding is initially starting up.
+ */
+ if (GetActiveWalLevelOnStandby() < WAL_LEVEL_LOGICAL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("logical decoding on standby requires "
+ "wal_level >= logical on master")));
+ }
}
/*
@@ -330,6 +329,12 @@ CreateInitDecodingContext(const char *plugin,
LogicalDecodingContext *ctx;
MemoryContext old_context;
+ /*
+ * On standby, this check is also required while creating the slot. Check
+ * the comments in this function.
+ */
+ CheckLogicalDecodingRequirements();
+
/* shorter lines... */
slot = MyReplicationSlot;
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index 805d4c5a5b..7d8890a22a 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -214,7 +214,7 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
if (!RecoveryInProgress())
end_of_wal = GetFlushRecPtr();
else
- end_of_wal = GetXLogReplayRecPtr(&ThisTimeLineID);
+ end_of_wal = GetXLogReplayRecPtr(&ThisTimeLineID, false);
ReplicationSlotAcquire(NameStr(*name), true);
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 188716957f..81a3e72732 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -1100,37 +1100,28 @@ ReplicationSlotReserveWal(void)
/*
* For logical slots log a standby snapshot and start logical decoding
* at exactly that position. That allows the slot to start up more
- * quickly.
+ * quickly. But on a standby we cannot do WAL writes, so just use the
+ * replay pointer; effectively, an attempt to create a logical slot on
+ * standby will cause it to wait for an xl_running_xact record to be
+ * logged independently on the primary, so that a snapshot can be built
+ * using the record.
*
- * That's not needed (or indeed helpful) for physical slots as they'll
- * start replay at the last logged checkpoint anyway. Instead return
- * the location of the last redo LSN. While that slightly increases
- * the chance that we have to retry, it's where a base backup has to
- * start replay at.
+ * None of this is needed (or indeed helpful) for physical slots as
+ * they'll start replay at the last logged checkpoint anyway. Instead
+ * return the location of the last redo LSN. While that slightly
+ * increases the chance that we have to retry, it's where a base backup
+ * has to start replay at.
*/
- if (!RecoveryInProgress() && SlotIsLogical(slot))
- {
- XLogRecPtr flushptr;
-
- /* start at current insert position */
+ if (SlotIsPhysical(slot))
+ restart_lsn = GetRedoRecPtr();
+ else if (RecoveryInProgress())
+ restart_lsn = GetXLogReplayRecPtr(NULL, true);
+ else
restart_lsn = GetXLogInsertRecPtr();
- SpinLockAcquire(&slot->mutex);
- slot->data.restart_lsn = restart_lsn;
- SpinLockRelease(&slot->mutex);
-
- /* make sure we have enough information to start */
- flushptr = LogStandbySnapshot();
- /* and make sure it's fsynced to disk */
- XLogFlush(flushptr);
- }
- else
- {
- restart_lsn = GetRedoRecPtr();
- SpinLockAcquire(&slot->mutex);
- slot->data.restart_lsn = restart_lsn;
- SpinLockRelease(&slot->mutex);
- }
+ SpinLockAcquire(&slot->mutex);
+ slot->data.restart_lsn = restart_lsn;
+ SpinLockRelease(&slot->mutex);
/* prevent WAL removal as fast as possible */
ReplicationSlotsComputeRequiredLSN();
@@ -1146,6 +1137,17 @@ ReplicationSlotReserveWal(void)
if (XLogGetLastRemovedSegno() < segno)
break;
}
+
+ if (!RecoveryInProgress() && SlotIsLogical(slot))
+ {
+ XLogRecPtr flushptr;
+
+ /* make sure we have enough information to start */
+ flushptr = LogStandbySnapshot();
+
+ /* and make sure it's fsynced to disk */
+ XLogFlush(flushptr);
+ }
}
/*
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 17df99c2ac..948acb1ccb 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -627,7 +627,7 @@ pg_replication_slot_advance(PG_FUNCTION_ARGS)
if (!RecoveryInProgress())
moveto = Min(moveto, GetFlushRecPtr());
else
- moveto = Min(moveto, GetXLogReplayRecPtr(&ThisTimeLineID));
+ moveto = Min(moveto, GetXLogReplayRecPtr(&ThisTimeLineID, false));
/* Acquire the slot so we "own" it */
ReplicationSlotAcquire(NameStr(*slotname), true);
diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index b90e5ca98e..28cc123c09 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -408,7 +408,7 @@ WalReceiverMain(void)
first_stream = false;
/* Initialize LogstreamResult and buffers for processing messages */
- LogstreamResult.Write = LogstreamResult.Flush = GetXLogReplayRecPtr(NULL);
+ LogstreamResult.Write = LogstreamResult.Flush = GetXLogReplayRecPtr(NULL, false);
initStringInfo(&reply_message);
initStringInfo(&incoming_message);
@@ -1098,7 +1098,7 @@ XLogWalRcvSendReply(bool force, bool requestReply)
/* Construct a new message */
writePtr = LogstreamResult.Write;
flushPtr = LogstreamResult.Flush;
- applyPtr = GetXLogReplayRecPtr(NULL);
+ applyPtr = GetXLogReplayRecPtr(NULL, false);
resetStringInfo(&reply_message);
pq_sendbyte(&reply_message, 'r');
diff --git a/src/backend/replication/walreceiverfuncs.c b/src/backend/replication/walreceiverfuncs.c
index 6f0acbfdef..f264b71f73 100644
--- a/src/backend/replication/walreceiverfuncs.c
+++ b/src/backend/replication/walreceiverfuncs.c
@@ -372,7 +372,7 @@ GetReplicationApplyDelay(void)
receivePtr = walrcv->flushedUpto;
SpinLockRelease(&walrcv->mutex);
- replayPtr = GetXLogReplayRecPtr(NULL);
+ replayPtr = GetXLogReplayRecPtr(NULL, false);
if (receivePtr == replayPtr)
return 0;
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 63c0dd2f62..2ea4d27b63 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -535,7 +535,7 @@ ReadReplicationSlot(ReadReplicationSlotCmd *cmd)
* to get the LSN position's history.
*/
if (RecoveryInProgress())
- (void) GetXLogReplayRecPtr(¤t_timeline);
+ (void) GetXLogReplayRecPtr(¤t_timeline, false);
else
current_timeline = ThisTimeLineID;
@@ -1273,6 +1273,16 @@ StartLogicalReplication(StartReplicationCmd *cmd)
got_STOPPING = true;
}
+ /*
+ * In case of logical decoding on standby it may be that ThisTimeLineID
+ * is not set yet.
+ * Indeed we are not going through InitXLOGAccess on a Standby and
+ * it may also be that IdentifySystem has not been called yet.
+ * So let's get it through GetXLogReplayRecPtr().
+ */
+ if (ThisTimeLineID == 0)
+ (void) GetXLogReplayRecPtr(&ThisTimeLineID, false);
+
/*
* Create our decoding context, making it start at the previously ack'ed
* position.
@@ -1497,7 +1507,7 @@ WalSndWaitForWal(XLogRecPtr loc)
if (!RecoveryInProgress())
RecentFlushPtr = GetFlushRecPtr();
else
- RecentFlushPtr = GetXLogReplayRecPtr(NULL);
+ RecentFlushPtr = GetXLogReplayRecPtr(NULL, false);
for (;;)
{
@@ -1531,7 +1541,7 @@ WalSndWaitForWal(XLogRecPtr loc)
if (!RecoveryInProgress())
RecentFlushPtr = GetFlushRecPtr();
else
- RecentFlushPtr = GetXLogReplayRecPtr(NULL);
+ RecentFlushPtr = GetXLogReplayRecPtr(NULL, false);
/*
* If postmaster asked us to stop, don't wait anymore.
@@ -3004,10 +3014,12 @@ XLogSendLogical(void)
* If first time through in this session, initialize flushPtr. Otherwise,
* we only need to update flushPtr if EndRecPtr is past it.
*/
- if (flushPtr == InvalidXLogRecPtr)
- flushPtr = GetFlushRecPtr();
- else if (logical_decoding_ctx->reader->EndRecPtr >= flushPtr)
- flushPtr = GetFlushRecPtr();
+ if (flushPtr == InvalidXLogRecPtr ||
+ logical_decoding_ctx->reader->EndRecPtr >= flushPtr)
+ {
+ flushPtr = (am_cascading_walsender ?
+ GetStandbyFlushRecPtr() : GetFlushRecPtr());
+ }
/* If EndRecPtr is still past our flushPtr, it means we caught up. */
if (logical_decoding_ctx->reader->EndRecPtr >= flushPtr)
@@ -3096,7 +3108,7 @@ GetStandbyFlushRecPtr(void)
*/
receivePtr = GetWalRcvFlushRecPtr(NULL, &receiveTLI);
- replayPtr = GetXLogReplayRecPtr(&replayTLI);
+ replayPtr = GetXLogReplayRecPtr(&replayTLI, false);
ThisTimeLineID = replayTLI;
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 5e2c94a05f..97061253e6 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -282,7 +282,7 @@ extern bool HotStandbyActive(void);
extern bool HotStandbyActiveInReplay(void);
extern bool XLogInsertAllowed(void);
extern void GetXLogReceiptTime(TimestampTz *rtime, bool *fromStream);
-extern XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI);
+extern XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI, bool avoid_header);
extern XLogRecPtr GetXLogInsertRecPtr(void);
extern XLogRecPtr GetXLogWriteRecPtr(void);
extern RecoveryPauseState GetRecoveryPauseState(void);
@@ -299,6 +299,7 @@ extern Size XLOGShmemSize(void);
extern void XLOGShmemInit(void);
extern void BootStrapXLOG(void);
extern void LocalProcessControlFile(bool reset);
+extern WalLevel GetActiveWalLevelOnStandby(void);
extern void StartupXLOG(void);
extern void ShutdownXLOG(int code, Datum arg);
extern void InitXLOGAccess(void);
--
2.18.4
[text/plain] v25-0004-New-TAP-test-for-logical-decoding-on-standby.patch (20.0K, ../[email protected]/6-v25-0004-New-TAP-test-for-logical-decoding-on-standby.patch)
download | inline diff:
From 873882b8dad1969124016785bc61243a658632a7 Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 26 Oct 2021 14:33:52 +0000
Subject: [PATCH v25 4/6] New TAP test for logical decoding on standby.
Author: Craig Ringer (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
src/test/perl/PostgreSQL/Test/Cluster.pm | 37 ++
.../t/027_standby_logical_decoding.pl | 498 ++++++++++++++++++
2 files changed, 535 insertions(+)
6.0% src/test/perl/PostgreSQL/Test/
93.9% src/test/recovery/t/
diff --git a/src/test/perl/PostgreSQL/Test/Cluster.pm b/src/test/perl/PostgreSQL/Test/Cluster.pm
index 86eb920ea1..89e49facb4 100644
--- a/src/test/perl/PostgreSQL/Test/Cluster.pm
+++ b/src/test/perl/PostgreSQL/Test/Cluster.pm
@@ -2804,6 +2804,43 @@ sub pg_recvlogical_upto
=pod
+=item $node->create_logical_slot_on_standby(self, master, slot_name, dbname)
+
+Create logical replication slot on given standby
+
+=cut
+
+sub create_logical_slot_on_standby
+{
+ my ($self, $master, $slot_name, $dbname) = @_;
+ my ($stdout, $stderr);
+
+ my $handle;
+
+ $handle = IPC::Run::start(['pg_recvlogical', '-d', $self->connstr($dbname), '-P', 'test_decoding', '-S', $slot_name, '--create-slot'], '>', \$stdout, '2>', \$stderr);
+
+ # Once slot restart_lsn is created, the standby looks for xl_running_xacts
+ # WAL record from the restart_lsn onwards. So firstly, wait until the slot
+ # restart_lsn is evaluated.
+
+ $self->poll_query_until(
+ 'postgres', qq[
+ SELECT restart_lsn IS NOT NULL
+ FROM pg_catalog.pg_replication_slots WHERE slot_name = '$slot_name'
+ ]) or die "timed out waiting for logical slot to calculate its restart_lsn";
+
+ # Now arrange for the xl_running_xacts record for which pg_recvlogical
+ # is waiting.
+ $master->safe_psql('postgres', 'CHECKPOINT');
+
+ $handle->finish();
+
+ is($self->slot($slot_name)->{'slot_type'}, 'logical', $slot_name . ' on standby created')
+ or die "could not create slot" . $slot_name;
+}
+
+=pod
+
=back
=cut
diff --git a/src/test/recovery/t/027_standby_logical_decoding.pl b/src/test/recovery/t/027_standby_logical_decoding.pl
new file mode 100644
index 0000000000..cb75f2b0c3
--- /dev/null
+++ b/src/test/recovery/t/027_standby_logical_decoding.pl
@@ -0,0 +1,498 @@
+# logical decoding on standby : test logical decoding,
+# recovery conflict and standby promotion.
+
+use strict;
+use warnings;
+
+use PostgreSQL::Test::Cluster;
+use Test::More tests => 38;
+use Time::HiRes qw(usleep);
+
+my ($stdin, $stdout, $stderr, $ret, $handle, $slot);
+
+my $node_primary = PostgreSQL::Test::Cluster->new('primary');
+my $node_standby = PostgreSQL::Test::Cluster->new('standby');
+
+# Name for the physical slot on primary
+my $primary_slotname = 'primary_physical';
+
+# return the size of logfile of $node in bytes
+sub get_log_size
+{
+ my ($node) = @_;
+
+ return (stat $node->logfile)[7];
+}
+
+# find $pat in logfile of $node after $off-th byte
+sub find_in_log
+{
+ my ($node, $pat, $off) = @_;
+
+ $off = 0 unless defined $off;
+ my $log = PostgreSQL::Test::Utils::slurp_file($node->logfile);
+ return 0 if (length($log) <= $off);
+
+ $log = substr($log, $off);
+
+ return $log =~ m/$pat/;
+}
+
+# Fetch xmin columns from slot's pg_replication_slots row, after waiting for
+# given boolean condition to be true to ensure we've reached a quiescent state.
+sub wait_for_xmins
+{
+ my ($node, $slotname, $check_expr) = @_;
+
+ $node->poll_query_until(
+ 'postgres', qq[
+ SELECT $check_expr
+ FROM pg_catalog.pg_replication_slots
+ WHERE slot_name = '$slotname';
+ ]) or die "Timed out waiting for slot xmins to advance";
+}
+
+# Create the required logical slots on standby.
+sub create_logical_slots
+{
+ $node_standby->create_logical_slot_on_standby($node_primary, 'inactiveslot', 'testdb');
+ $node_standby->create_logical_slot_on_standby($node_primary, 'activeslot', 'testdb');
+}
+
+# Acquire one of the standby logical slots created by create_logical_slots().
+# In case wait is true we are waiting for an active pid on the 'activeslot' slot.
+# If wait is not true it means we are testing a known failure scenario.
+sub make_slot_active
+{
+ my $wait = shift;
+ my $slot_user_handle;
+
+ print "starting pg_recvlogical\n";
+ $slot_user_handle = IPC::Run::start(['pg_recvlogical', '-d', $node_standby->connstr('testdb'), '-S', 'activeslot', '-f', '-', '--no-loop', '--start'], '>', \$stdout, '2>', \$stderr);
+
+ if ($wait)
+ # make sure activeslot is in use
+ {
+ $node_standby->poll_query_until('testdb',
+ "SELECT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = 'activeslot' AND active_pid IS NOT NULL)"
+ ) or die "slot never became active";
+ }
+
+ return $slot_user_handle;
+}
+
+# Check if all the slots on standby are dropped. These include the 'activeslot'
+# that was acquired by make_slot_active(), and the non-active 'inactiveslot'.
+sub check_slots_dropped
+{
+ my ($slot_user_handle) = @_;
+ my $return;
+
+ is($node_standby->slot('inactiveslot')->{'slot_type'}, '', 'inactiveslot on standby dropped');
+ is($node_standby->slot('activeslot')->{'slot_type'}, '', 'activeslot on standby dropped');
+
+ # our client should've terminated in response to the walsender error
+ $slot_user_handle->finish;
+ $return = $?;
+ cmp_ok($return, "!=", 0, "pg_recvlogical exited non-zero");
+ if ($return) {
+ like($stderr, qr/conflict with recovery/, 'slot have been dropped');
+ }
+
+ return 0;
+}
+
+
+########################
+# Initialize primary node
+########################
+
+$node_primary->init(allows_streaming => 1, has_archiving => 1);
+$node_primary->append_conf('postgresql.conf', q{
+wal_level = 'logical'
+max_replication_slots = 4
+max_wal_senders = 4
+log_min_messages = 'debug2'
+log_error_verbosity = verbose
+});
+$node_primary->dump_info;
+$node_primary->start;
+
+$node_primary->psql('postgres', q[CREATE DATABASE testdb]);
+
+$node_primary->safe_psql('testdb', qq[SELECT * FROM pg_create_physical_replication_slot('$primary_slotname');]);
+my $backup_name = 'b1';
+$node_primary->backup($backup_name);
+
+#######################
+# Initialize standby node
+#######################
+
+$node_standby->init_from_backup(
+ $node_primary, $backup_name,
+ has_streaming => 1,
+ has_restoring => 1);
+$node_standby->append_conf('postgresql.conf',
+ qq[primary_slot_name = '$primary_slotname']);
+$node_standby->start;
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+
+##################################################
+# Test that logical decoding on the standby
+# behaves correctly.
+##################################################
+
+create_logical_slots();
+
+$node_primary->safe_psql('testdb', qq[CREATE TABLE decoding_test(x integer, y text);]);
+$node_primary->safe_psql('testdb', qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(1,10) s;]);
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+my $result = $node_standby->safe_psql('testdb',
+ qq[SELECT pg_logical_slot_get_changes('activeslot', NULL, NULL);]);
+
+# test if basic decoding works
+is(scalar(my @foobar = split /^/m, $result),
+ 14, 'Decoding produced 14 rows');
+
+# Insert some rows and verify that we get the same results from pg_recvlogical
+# and the SQL interface.
+$node_primary->safe_psql('testdb',
+ qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(1,4) s;]
+);
+
+my $expected = q{BEGIN
+table public.decoding_test: INSERT: x[integer]:1 y[text]:'1'
+table public.decoding_test: INSERT: x[integer]:2 y[text]:'2'
+table public.decoding_test: INSERT: x[integer]:3 y[text]:'3'
+table public.decoding_test: INSERT: x[integer]:4 y[text]:'4'
+COMMIT};
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+my $stdout_sql = $node_standby->safe_psql('testdb',
+ qq[SELECT data FROM pg_logical_slot_peek_changes('activeslot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');]
+);
+
+is($stdout_sql, $expected, 'got expected output from SQL decoding session');
+
+my $endpos = $node_standby->safe_psql('testdb',
+ "SELECT lsn FROM pg_logical_slot_peek_changes('activeslot', NULL, NULL) ORDER BY lsn DESC LIMIT 1;"
+);
+print "waiting to replay $endpos\n";
+
+# Insert some rows after $endpos, which we won't read.
+$node_primary->safe_psql('testdb',
+ qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(5,50) s;]
+);
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+my $stdout_recv = $node_standby->pg_recvlogical_upto(
+ 'testdb', 'activeslot', $endpos, 180,
+ 'include-xids' => '0',
+ 'skip-empty-xacts' => '1');
+chomp($stdout_recv);
+is($stdout_recv, $expected,
+ 'got same expected output from pg_recvlogical decoding session');
+
+$node_standby->poll_query_until('testdb',
+ "SELECT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = 'activeslot' AND active_pid IS NULL)"
+) or die "slot never became inactive";
+
+$stdout_recv = $node_standby->pg_recvlogical_upto(
+ 'testdb', 'activeslot', $endpos, 180,
+ 'include-xids' => '0',
+ 'skip-empty-xacts' => '1');
+chomp($stdout_recv);
+is($stdout_recv, '', 'pg_recvlogical acknowledged changes');
+
+$node_primary->safe_psql('postgres', 'CREATE DATABASE otherdb');
+
+is( $node_primary->psql(
+ 'otherdb',
+ "SELECT lsn FROM pg_logical_slot_peek_changes('activeslot', NULL, NULL) ORDER BY lsn DESC LIMIT 1;"
+ ),
+ 3,
+ 'replaying logical slot from another database fails');
+
+# drop the logical slots
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('inactiveslot')]);
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('activeslot')]);
+
+##################################################
+# Recovery conflict: Invalidate conflicting slots, including in-use slots
+# Scenario 1: hot_standby_feedback off and vacuum FULL
+##################################################
+
+create_logical_slots();
+
+# One way to reproduce recovery conflict is to run VACUUM FULL with
+# hot_standby_feedback turned off on the standby.
+$node_standby->append_conf('postgresql.conf',q[
+hot_standby_feedback = off
+]);
+$node_standby->restart;
+# ensure walreceiver feedback off by waiting for expected xmin and
+# catalog_xmin on primary. Both should be NULL since hs_feedback is off
+wait_for_xmins($node_primary, $primary_slotname,
+ "xmin IS NULL AND catalog_xmin IS NULL");
+
+$handle = make_slot_active(1);
+
+# This should trigger the conflict
+$node_primary->safe_psql('testdb', 'VACUUM FULL');
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+# message should be issued
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"inactiveslot\" because it conflicts with recovery"),
+ 'inactiveslot slot invalidation is logged with vacuum FULL');
+
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"activeslot\" because it conflicts with recovery"),
+ 'activeslot slot invalidation is logged with vacuum FULL');
+
+# Verify that pg_stat_database_conflicts.confl_logicalslot has been updated
+ok( $node_standby->poll_query_until(
+ 'postgres',
+ "select (confl_logicalslot = 2) from pg_stat_database_conflicts where datname = 'testdb'", 't'),
+ 'confl_logicalslot updated') or die "Timed out waiting confl_logicalslot to be updated";
+
+$handle = make_slot_active(0);
+usleep(100_000);
+
+# We are not able to read from the slot as it has been invalidated
+ok( find_in_log(
+ $node_standby,
+ "cannot read from logical replication slot \"activeslot\""),
+ 'cannot read from logical replication slot');
+
+# Turn hot_standby_feedback back on
+$node_standby->append_conf('postgresql.conf',q[
+hot_standby_feedback = on
+]);
+$node_standby->restart;
+
+# ensure walreceiver feedback sent by waiting for expected xmin and
+# catalog_xmin on primary. With hot_standby_feedback on, xmin should advance,
+# but catalog_xmin should still remain NULL since there is no logical slot.
+wait_for_xmins($node_primary, $primary_slotname,
+ "xmin IS NOT NULL AND catalog_xmin IS NULL");
+
+##################################################
+# Recovery conflict: Invalidate conflicting slots, including in-use slots
+# Scenario 2: conflict due to row removal with hot_standby_feedback off.
+##################################################
+
+# get the position to search from in the standby logfile
+my $logstart = get_log_size($node_standby);
+
+# drop the logical slots
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('inactiveslot')]);
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('activeslot')]);
+
+create_logical_slots();
+
+# One way to produce recovery conflict is to create/drop a relation and launch a vacuum
+# with hot_standby_feedback turned off on the standby.
+$node_standby->append_conf('postgresql.conf',q[
+hot_standby_feedback = off
+]);
+$node_standby->restart;
+# ensure walreceiver feedback off by waiting for expected xmin and
+# catalog_xmin on primary. Both should be NULL since hs_feedback is off
+wait_for_xmins($node_primary, $primary_slotname,
+ "xmin IS NULL AND catalog_xmin IS NULL");
+
+$handle = make_slot_active(1);
+
+# This should trigger the conflict
+$node_primary->safe_psql('testdb', qq[CREATE TABLE conflict_test(x integer, y text);]);
+$node_primary->safe_psql('testdb', qq[DROP TABLE conflict_test;]);
+$node_primary->safe_psql('testdb', 'VACUUM');
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+# message should be issued
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"inactiveslot\" because it conflicts with recovery", $logstart),
+ 'inactiveslot slot invalidation is logged due to row removal');
+
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"activeslot\" because it conflicts with recovery", $logstart),
+ 'activeslot slot invalidation is logged due to row removal');
+
+# Verify that pg_stat_database_conflicts.confl_logicalslot has been updated
+ok( $node_standby->poll_query_until(
+ 'postgres',
+ "select (confl_logicalslot = 2) from pg_stat_database_conflicts where datname = 'testdb'", 't'),
+ 'confl_logicalslot updated') or die "Timed out waiting confl_logicalslot to be updated";
+
+$handle = make_slot_active(0);
+usleep(100_000);
+
+# We are not able to read from the slot as it has been invalidated
+ok( find_in_log(
+ $node_standby,
+ "cannot read from logical replication slot \"activeslot\"", $logstart),
+ 'cannot read from logical replication slot');
+
+# Turn hot_standby_feedback back on
+$node_standby->append_conf('postgresql.conf',q[
+hot_standby_feedback = on
+]);
+$node_standby->restart;
+
+# ensure walreceiver feedback sent by waiting for expected xmin and
+# catalog_xmin on primary. With hot_standby_feedback on, xmin should advance,
+# but catalog_xmin should still remain NULL since there is no logical slot.
+wait_for_xmins($node_primary, $primary_slotname,
+ "xmin IS NOT NULL AND catalog_xmin IS NULL");
+
+##################################################
+# Recovery conflict: Invalidate conflicting slots, including in-use slots
+# Scenario 3: incorrect wal_level on primary.
+##################################################
+
+# get the position to search from in the standby logfile
+$logstart = get_log_size($node_standby);
+
+# drop the logical slots
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('inactiveslot')]);
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('activeslot')]);
+
+create_logical_slots();
+
+$handle = make_slot_active(1);
+
+# Make primary wal_level replica. This will trigger slot conflict.
+$node_primary->append_conf('postgresql.conf',q[
+wal_level = 'replica'
+]);
+$node_primary->restart;
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+# message should be issued
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"inactiveslot\" because it conflicts with recovery", $logstart),
+ 'inactiveslot slot invalidation is logged due to wal_level');
+
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"activeslot\" because it conflicts with recovery", $logstart),
+ 'activeslot slot invalidation is logged due to wal_level');
+
+# Verify that pg_stat_database_conflicts.confl_logicalslot has been updated
+ok( $node_standby->poll_query_until(
+ 'postgres',
+ "select (confl_logicalslot = 2) from pg_stat_database_conflicts where datname = 'testdb'", 't'),
+ 'confl_logicalslot updated') or die "Timed out waiting confl_logicalslot to be updated";
+
+$handle = make_slot_active(0);
+usleep(100_000);
+
+ok( find_in_log(
+ $node_standby,
+ "logical decoding on standby requires wal_level >= logical on master", $logstart),
+ 'cannot start replication because wal_level < logical on master');
+
+# Restore primary wal_level
+$node_primary->append_conf('postgresql.conf',q[
+wal_level = 'logical'
+]);
+$node_primary->restart;
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+$handle = make_slot_active(0);
+usleep(100_000);
+
+# as the slot has been invalidated we should not be able to read
+ok( find_in_log(
+ $node_standby,
+ "cannot read from logical replication slot \"activeslot\"", $logstart),
+ 'cannot read from logical replication slot');
+
+##################################################
+# DROP DATABASE should drops it's slots, including active slots.
+##################################################
+
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('inactiveslot')]);
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('activeslot')]);
+create_logical_slots();
+$handle = make_slot_active(1);
+# Create a slot on a database that would not be dropped. This slot should not
+# get dropped.
+$node_standby->create_logical_slot_on_standby($node_primary, 'otherslot', 'postgres');
+
+# dropdb on the primary to verify slots are dropped on standby
+$node_primary->safe_psql('postgres', q[DROP DATABASE testdb]);
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+is($node_standby->safe_psql('postgres',
+ q[SELECT EXISTS(SELECT 1 FROM pg_database WHERE datname = 'testdb')]), 'f',
+ 'database dropped on standby');
+
+check_slots_dropped($handle);
+
+is($node_standby->slot('otherslot')->{'slot_type'}, 'logical',
+ 'otherslot on standby not dropped');
+
+# Cleanup : manually drop the slot that was not dropped.
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('otherslot')]);
+
+##################################################
+# Test standby promotion and logical decoding behavior
+# after the standby gets promoted.
+##################################################
+
+$node_primary->psql('postgres', q[CREATE DATABASE testdb]);
+$node_primary->safe_psql('testdb', qq[CREATE TABLE decoding_test(x integer, y text);]);
+
+# create the logical slots
+create_logical_slots();
+
+# Insert some rows before the promotion
+$node_primary->safe_psql('testdb',
+ qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(1,4) s;]
+);
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+# promote
+$node_standby->promote;
+
+# insert some rows on promoted standby
+$node_standby->safe_psql('testdb',
+ qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(5,7) s;]
+);
+
+
+$expected = q{BEGIN
+table public.decoding_test: INSERT: x[integer]:1 y[text]:'1'
+table public.decoding_test: INSERT: x[integer]:2 y[text]:'2'
+table public.decoding_test: INSERT: x[integer]:3 y[text]:'3'
+table public.decoding_test: INSERT: x[integer]:4 y[text]:'4'
+COMMIT
+BEGIN
+table public.decoding_test: INSERT: x[integer]:5 y[text]:'5'
+table public.decoding_test: INSERT: x[integer]:6 y[text]:'6'
+table public.decoding_test: INSERT: x[integer]:7 y[text]:'7'
+COMMIT};
+
+# check that we are decoding pre and post promotion inserted rows
+$stdout_sql = $node_standby->safe_psql('testdb',
+ qq[SELECT data FROM pg_logical_slot_peek_changes('activeslot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');]
+);
+
+is($stdout_sql, $expected, 'got expected output from SQL decoding session on promoted standby');
--
2.18.4
[text/plain] v25-0005-Doc-changes-describing-details-about-logical-dec.patch (2.1K, ../[email protected]/7-v25-0005-Doc-changes-describing-details-about-logical-dec.patch)
download | inline diff:
From 7b10524f2e81e4a0beb160e6297cfb6d601b6d64 Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 26 Oct 2021 14:34:37 +0000
Subject: [PATCH v25 5/6] Doc changes describing details about logical
decoding.
Author: Andres Freund (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
doc/src/sgml/logicaldecoding.sgml | 21 +++++++++++++++++++++
1 file changed, 21 insertions(+)
100.0% doc/src/sgml/
diff --git a/doc/src/sgml/logicaldecoding.sgml b/doc/src/sgml/logicaldecoding.sgml
index b6353c7a12..98fffc4352 100644
--- a/doc/src/sgml/logicaldecoding.sgml
+++ b/doc/src/sgml/logicaldecoding.sgml
@@ -316,6 +316,27 @@ postgres=# select * from pg_logical_slot_get_changes('regression_slot', NULL, NU
may consume changes from a slot at any given time.
</para>
+ <para>
+ A logical replication slot can also be created on a hot standby. To prevent
+ <command>VACUUM</command> from removing required rows from the system
+ catalogs, <varname>hot_standby_feedback</varname> should be set on the
+ standby. In spite of that, if any required rows get removed, the slot gets
+ invalidated. It's highly recommended to use a physical slot between the primary
+ and the standby. Otherwise, hot_standby_feedback will work, but only while the
+ connection is alive (for example a node restart would break it). Existing
+ logical slots on standby also get invalidated if wal_level on primary is reduced to
+ less than 'logical'.
+ </para>
+
+ <para>
+ For a logical slot to be created, it builds a historic snapshot, for which
+ information of all the currently running transactions is essential. On
+ primary, this information is available, but on standby, this information
+ has to be obtained from primary. So, slot creation may wait for some
+ activity to happen on the primary. If the primary is idle, creating a
+ logical slot on standby may take a noticeable time.
+ </para>
+
<caution>
<para>
Replication slots persist across crashes and know nothing about the state
--
2.18.4
[text/plain] v25-0006-Fixing-Walsender-corner-cases-with-logical-decod.patch (7.2K, ../[email protected]/8-v25-0006-Fixing-Walsender-corner-cases-with-logical-decod.patch)
download | inline diff:
From 95cdd939b709a2a1f1462b6a92c6cc201a3adde7 Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 26 Oct 2021 14:35:23 +0000
Subject: [PATCH v25 6/6] Fixing Walsender corner cases with logical decoding
on standby.
The problem is that WalSndWaitForWal() waits for the *replay* LSN to
increase, but gets woken up by walreceiver when new WAL has been
flushed. Which means that typically walsenders will get woken up at the
same time that the startup process will be - which means that by the
time the logical walsender checks GetXLogReplayRecPtr() it's unlikely
that the startup process already replayed the record and updated
XLogCtl->lastReplayedEndRecPtr.
Fixed by making used of a condition variable on the replay position.
---
src/backend/access/transam/xlog.c | 20 +++++++++++++----
src/backend/replication/walsender.c | 29 ++++++++++++++++++-------
src/backend/utils/activity/wait_event.c | 3 +++
src/include/replication/walsender.h | 12 ++++++++++
src/include/utils/wait_event.h | 1 +
5 files changed, 53 insertions(+), 12 deletions(-)
29.6% src/backend/access/transam/
52.7% src/backend/replication/
4.1% src/backend/utils/activity/
11.7% src/include/replication/
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 1e9f94927c..3d125228fe 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -737,6 +737,7 @@ typedef struct XLogCtlData
} XLogCtlData;
static XLogCtlData *XLogCtl = NULL;
+XLogCtlCvData *XLogCtlCv = NULL;
/* a private copy of XLogCtl->Insert.WALInsertLocks, for convenience */
static WALInsertLockPadded *WALInsertLocks = NULL;
@@ -5182,7 +5183,8 @@ void
XLOGShmemInit(void)
{
bool foundCFile,
- foundXLog;
+ foundXLog,
+ foundXLogCv;
char *allocptr;
int i;
ControlFileData *localControlFile;
@@ -5207,14 +5209,17 @@ XLOGShmemInit(void)
XLogCtl = (XLogCtlData *)
ShmemInitStruct("XLOG Ctl", XLOGShmemSize(), &foundXLog);
+ XLogCtlCv = (XLogCtlCvData *)
+ ShmemInitStruct("XLOG Cv Ctl", sizeof(XLogCtlCvData), &foundXLogCv);
+
localControlFile = ControlFile;
ControlFile = (ControlFileData *)
ShmemInitStruct("Control File", sizeof(ControlFileData), &foundCFile);
- if (foundCFile || foundXLog)
+ if (foundCFile || foundXLog || foundXLogCv)
{
- /* both should be present or neither */
- Assert(foundCFile && foundXLog);
+ /* All should be present or neither */
+ Assert(foundCFile && foundXLog && foundXLogCv);
/* Initialize local copy of WALInsertLocks */
WALInsertLocks = XLogCtl->Insert.WALInsertLocks;
@@ -5224,6 +5229,7 @@ XLOGShmemInit(void)
return;
}
memset(XLogCtl, 0, sizeof(XLogCtlData));
+ memset(XLogCtlCv, 0, sizeof(XLogCtlCvData));
/*
* Already have read control file locally, unless in bootstrap mode. Move
@@ -5285,6 +5291,7 @@ XLOGShmemInit(void)
SpinLockInit(&XLogCtl->ulsn_lck);
InitSharedLatch(&XLogCtl->recoveryWakeupLatch);
ConditionVariableInit(&XLogCtl->recoveryNotPausedCV);
+ ConditionVariableInit(&XLogCtlCv->replayedCV);
}
/*
@@ -7684,6 +7691,11 @@ StartupXLOG(void)
XLogCtl->lastReplayedTLI = ThisTimeLineID;
SpinLockRelease(&XLogCtl->info_lck);
+ /*
+ * wake up walsender(s) used by logical decoding on standby.
+ */
+ ConditionVariableBroadcast(&XLogCtlCv->replayedCV);
+
/*
* If rm_redo called XLogRequestWalReceiverReply, then we wake
* up the receiver so that it notices the updated
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 2ea4d27b63..8460413030 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1493,6 +1493,7 @@ WalSndWaitForWal(XLogRecPtr loc)
{
int wakeEvents;
static XLogRecPtr RecentFlushPtr = InvalidXLogRecPtr;
+ XLogCtlCvData *xlogctlcv = XLogCtlCv;
/*
* Fast path to avoid acquiring the spinlock in case we already know we
@@ -1511,7 +1512,6 @@ WalSndWaitForWal(XLogRecPtr loc)
for (;;)
{
- long sleeptime;
/* Clear any already-pending wakeups */
ResetLatch(MyLatch);
@@ -1595,20 +1595,33 @@ WalSndWaitForWal(XLogRecPtr loc)
WalSndKeepaliveIfNecessary();
/*
- * Sleep until something happens or we time out. Also wait for the
- * socket becoming writable, if there's still pending output.
+ * When not in recovery, sleep until something happens or we time out.
+ * Also wait for the socket becoming writable, if there's still pending output.
* Otherwise we might sit on sendable output data while waiting for
* new WAL to be generated. (But if we have nothing to send, we don't
* want to wake on socket-writable.)
*/
- sleeptime = WalSndComputeSleeptime(GetCurrentTimestamp());
+ if (!RecoveryInProgress())
+ {
+ long sleeptime;
+ sleeptime = WalSndComputeSleeptime(GetCurrentTimestamp());
- wakeEvents = WL_SOCKET_READABLE;
+ wakeEvents = WL_SOCKET_READABLE;
- if (pq_is_send_pending())
- wakeEvents |= WL_SOCKET_WRITEABLE;
+ if (pq_is_send_pending())
+ wakeEvents |= WL_SOCKET_WRITEABLE;
- WalSndWait(wakeEvents, sleeptime, WAIT_EVENT_WAL_SENDER_WAIT_WAL);
+ WalSndWait(wakeEvents, sleeptime * 10, WAIT_EVENT_WAL_SENDER_WAIT_WAL);
+ }
+ else
+ /*
+ * We are in the logical decoding on standby case.
+ * We are waiting for the startup process to replay wal record(s).
+ */
+ {
+ ConditionVariablePrepareToSleep(&xlogctlcv->replayedCV);
+ ConditionVariableSleep(&xlogctlcv->replayedCV, WAIT_EVENT_WAL_SENDER_WAIT_REPLAY);
+ }
}
/* reactivate latch so WalSndLoop knows to continue */
diff --git a/src/backend/utils/activity/wait_event.c b/src/backend/utils/activity/wait_event.c
index 4a5b7502f5..92dc17baf1 100644
--- a/src/backend/utils/activity/wait_event.c
+++ b/src/backend/utils/activity/wait_event.c
@@ -448,6 +448,9 @@ pgstat_get_wait_ipc(WaitEventIPC w)
case WAIT_EVENT_WAL_RECEIVER_WAIT_START:
event_name = "WalReceiverWaitStart";
break;
+ case WAIT_EVENT_WAL_SENDER_WAIT_REPLAY:
+ event_name = "WalReceiverWaitReplay";
+ break;
case WAIT_EVENT_XACT_GROUP_UPDATE:
event_name = "XactGroupUpdate";
break;
diff --git a/src/include/replication/walsender.h b/src/include/replication/walsender.h
index 828106933c..7a2c04c937 100644
--- a/src/include/replication/walsender.h
+++ b/src/include/replication/walsender.h
@@ -13,6 +13,7 @@
#define _WALSENDER_H
#include <signal.h>
+#include "storage/condition_variable.h"
/*
* What to do with a snapshot in create replication slot command.
@@ -48,6 +49,17 @@ extern void WalSndWaitStopping(void);
extern void HandleWalSndInitStopping(void);
extern void WalSndRqstFileReload(void);
+/*
+ * shared-memory state for Condition Variable(s)
+ * between the startup process and the walsender.
+ */
+typedef struct XLogCtlCvData
+{
+ ConditionVariable replayedCV;
+} XLogCtlCvData;
+
+extern XLogCtlCvData *XLogCtlCv;
+
/*
* Remember that we want to wakeup walsenders later
*
diff --git a/src/include/utils/wait_event.h b/src/include/utils/wait_event.h
index c22142365f..b1a27f5e84 100644
--- a/src/include/utils/wait_event.h
+++ b/src/include/utils/wait_event.h
@@ -125,6 +125,7 @@ typedef enum
WAIT_EVENT_SYNC_REP,
WAIT_EVENT_WAL_RECEIVER_EXIT,
WAIT_EVENT_WAL_RECEIVER_WAIT_START,
+ WAIT_EVENT_WAL_SENDER_WAIT_REPLAY,
WAIT_EVENT_XACT_GROUP_UPDATE
} WaitEventIPC;
--
2.18.4
view thread (196+ 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], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: Re: Minimal logical decoding on standbys
In-Reply-To: <[email protected]>
* 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