public inbox for [email protected]
help / color / mirror / Atom feedFrom: Oh, Mike <[email protected]>
To: [email protected] <[email protected]>
Cc: Drouvot, Bertrand <[email protected]>
Subject: [BUG] Logical replication failure "ERROR: could not map filenode "base/13237/442428" to relation OID" with catalog modifying txns
Date: Mon, 15 Mar 2021 16:34:56 +0000
Message-ID: <[email protected]> (raw)
In-Reply-To: <[email protected]>
References: <[email protected]>
Sending this to pgsql-hackers list to create a CommitFest entry with the attached patch proposal.
Hello,
We noticed that the logical replication could fail when the Standby::RUNNING_XACT record is generated in the middle of a catalog modifying transaction and if the logical decoding has to restart from the RUNNING_XACT
WAL entry.
The Standby::RUNNING_XACT record is generated periodically (roughly every 15s by default) or during a CHECKPOINT operation.
Detailed problem description:
Tested on 11.8 & current master.
The logical replication slot restart_lsn advances in the middle of an open txn that modified the catalog (e.g. TRUNCATE operation).
Should the logical decoding has to restart it could fail with an error like this:
ERROR: could not map filenode "base/13237/442428"
Currently, the system relies on processing Heap2::NEW_CID to keep track of catalog modifying (sub)transactions.
This context is lost if the logical decoding has to restart from a Standby::RUNNING_XACTS that is written between the NEW_CID record and its parent txn commit.
If the logical stream restarts from this restart_lsn, then it doesn't have the xid responsible for modifying the catalog.
Repro steps:
1. We need to generate the Standby::RUNNING_XACT record deterministically using CHECKPOINT. Hence we'll delay the LOG_SNAPSHOT_INTERVAL_MS using the following patch:
diff --git a/src/backend/postmaster/bgwriter.c b/src/backend/postmaster/bgwriter.c
index 3e6ffb05b9..b776e8d566 100644
--- a/src/backend/postmaster/bgwriter.c
+++ b/src/backend/postmaster/bgwriter.c
@@ -76,7 +76,7 @@ int BgWriterDelay = 200;
* Interval in which standby snapshots are logged into the WAL stream, in
* milliseconds.
/
-#define LOG_SNAPSHOT_INTERVAL_MS 15000
+#define LOG_SNAPSHOT_INTERVAL_MS 1500000
2. Create a table
postgres=# create table bdt (a int);
CREATE TABLE
3. Create a logical replication slot:
postgres=# select pg_create_logical_replication_slot('bdt_slot','test_decoding');
pg_create_logical_replication_slot
------------------------------------
(bdt_slot,0/FFAA1C70)
(1 row)
4. Start reading the slot in a shell (keep the shell so that we can stop reading later):
./bin/pg_recvlogical --slot bdt_slot --start -f bdt.out -d postgres
5. Execute the workload across 2 different clients in the following order
Session1:
begin;
savepoint b1;
truncate bdt;
Session2:
select * from pg_replication_slots; /* keep note of the confirmed_flush_lsn */
checkpoint;
/* Repeat the following query until the confirmed_flush_lsn changes */
select * from pg_replication_slots;
Once confirmed_flush_lsn, changes:
Session1:
end;
begin;
insert into bdt values (1);
Session2:
select * from pg_replication_slots; /* keep note of both restart_lsn AND the confirmed_flush_lsn */
checkpoint;
/* Repeat the following query until both restart_lsn AND confirmed_flush_lsn change */
select * from pg_replication_slots;
6. Stop the pg_recvlogical (Control-C)
7. Then commit the insert txn:
Session1:
end;
8. Get/peek the replication slot changes
postgres=# select * from pg_logical_slot_get_changes('bdt_slot', null, null);
ERROR: could not map filenode "base/13237/442428" to relation OID
Proposed solution:
If we’re decoding a catalog modifying commit record, then check whether it’s part of the RUNNING_XACT xid’s processed @ the restart_lsn. If so, then add its xid & subxacts in the committed txns list in the logical decoding snapshot.
Please refer to the attachment for the proposed patch.
Thanks,
Mike
Attachments:
[application/octet-stream] logical_decoding_xact_bookkeep.patch (5.8K, ../[email protected]/3-logical_decoding_xact_bookkeep.patch)
download | inline diff:
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index 5f59613..2512912 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -42,6 +42,7 @@
#include "replication/reorderbuffer.h"
#include "replication/snapbuild.h"
#include "storage/standby.h"
+#include "utils/builtins.h"
typedef struct XLogRecordBuffer
{
@@ -85,6 +86,9 @@ static bool DecodeTXNNeedSkip(LogicalDecodingContext *ctx,
XLogRecordBuffer *buf, Oid dbId,
RepOriginId origin_id);
+/* record previous restart_lsn running xacts */
+xl_running_xacts *last_running = NULL;
+
/*
* Take every XLogReadRecord()ed record and perform the actions required to
* decode it using the output plugin already setup in the logical decoding
@@ -402,6 +406,28 @@ DecodeStandbyOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
{
xl_running_xacts *running = (xl_running_xacts *) XLogRecGetData(r);
+ /* record restart_lsn running xacts */
+ if (MyReplicationSlot && (buf->origptr == MyReplicationSlot->data.restart_lsn))
+ {
+ if (last_running)
+ free(last_running);
+
+ last_running = NULL;
+
+ /*
+ * xl_running_xacts contains a xids Flexible Array
+ * and its size is subxcnt + xcnt.
+ * Take that into account while allocating
+ * the memory for last_running.
+ */
+ last_running = (xl_running_xacts *) malloc(sizeof(xl_running_xacts)
+ + sizeof(TransactionId )
+ * (running->subxcnt + running->xcnt));
+ memcpy(last_running, running, sizeof(xl_running_xacts)
+ + (sizeof(TransactionId)
+ * (running->subxcnt + running->xcnt)));
+ }
+
SnapBuildProcessRunningXacts(builder, buf->origptr, running);
/*
@@ -678,6 +704,7 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
TimestampTz commit_time = parsed->xact_time;
RepOriginId origin_id = XLogRecGetOrigin(buf->record);
int i;
+ bool force_travel_and_snapshot = false;
if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
{
@@ -685,8 +712,29 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
commit_time = parsed->origin_timestamp;
}
+ /*
+ * Check if the commit contain catalog invalidations
+ * and if the xid was part of the restart_lsn
+ * running ones
+ */
+ if ((parsed->xinfo & XACT_XINFO_HAS_INVALS) && last_running)
+ {
+ /* make last_running->xids bsearch()able */
+ qsort(last_running->xids,
+ last_running->subxcnt + last_running->xcnt,
+ sizeof(TransactionId), xidComparator);
+
+ /*
+ * Is this xid part of the known running ones
+ * in the restart_lsn RUNNING_XACT entry?
+ */
+ force_travel_and_snapshot = TransactionIdInArray(xid, last_running->xids,
+ last_running->subxcnt
+ + last_running->xcnt);
+ }
+
SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid,
- parsed->nsubxacts, parsed->subxacts);
+ parsed->nsubxacts, parsed->subxacts, force_travel_and_snapshot);
/* ----
* Check whether we are interested in this specific transaction, and tell
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 91600ac..4a5610b 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -4892,7 +4892,7 @@ ApplyLogicalMappingFile(HTAB *tuplecid_data, Oid relid, const char *fname)
/*
* Check whether the TransactionId 'xid' is in the pre-sorted array 'xip'.
*/
-static bool
+bool
TransactionIdInArray(TransactionId xid, TransactionId *xip, Size num)
{
return bsearch(&xid, xip, num,
diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
index ed3acad..6be39d1 100644
--- a/src/backend/replication/logical/snapbuild.c
+++ b/src/backend/replication/logical/snapbuild.c
@@ -910,15 +910,19 @@ SnapBuildPurgeCommittedTxn(SnapBuild *builder)
*/
void
SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
- int nsubxacts, TransactionId *subxacts)
+ int nsubxacts, TransactionId *subxacts, bool force_travel_and_snapshot)
{
int nxact;
-
- bool needs_snapshot = false;
- bool needs_timetravel = false;
- bool sub_needs_timetravel = false;
+ bool needs_snapshot;
+ bool needs_timetravel;
+ bool sub_needs_timetravel;
TransactionId xmax = xid;
+ /*
+ * if force_travel_and_snapshot is set to true
+ * then proceed as if there are any catalog modifying subxacts.
+ */
+ needs_snapshot = needs_timetravel = sub_needs_timetravel = force_travel_and_snapshot;
/*
* Transactions preceding BUILDING_SNAPSHOT will neither be decoded, nor
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 565a961..7fc79f3 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -685,4 +685,6 @@ void ReorderBufferSetRestartPoint(ReorderBuffer *, XLogRecPtr ptr);
void StartupReorderBuffer(void);
+extern bool TransactionIdInArray(TransactionId xid, TransactionId *xip, Size num);
+
#endif
diff --git a/src/include/replication/snapbuild.h b/src/include/replication/snapbuild.h
index fbabce6..46f7df0 100644
--- a/src/include/replication/snapbuild.h
+++ b/src/include/replication/snapbuild.h
@@ -80,7 +80,7 @@ extern XLogRecPtr SnapBuildInitialConsistentPoint(SnapBuild *builder);
extern void SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn,
TransactionId xid, int nsubxacts,
- TransactionId *subxacts);
+ TransactionId *subxacts, bool force_travel_and_snapshot);
extern bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid,
XLogRecPtr lsn);
extern void SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid,
view thread (136+ 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]
Subject: Re: [BUG] Logical replication failure "ERROR: could not map filenode "base/13237/442428" to relation OID" with catalog modifying txns
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