v17-0004-Gracefully-handle-concurrent-aborts-of-uncommitt.patch

From ea6f204261c61d02cd9d956adff79053d4437a33 Mon Sep 17 00:00:00 2001
From: Dilip Kumar <dilip.kumar@enterprisedb.com>
Date: Thu, 9 Apr 2020 10:55:19 +0530
Subject: [PATCH v17 04/12] Gracefully handle concurrent aborts of uncommitted
 transactions that are being decoded alongside.

When a transaction aborts, it's changes are considered unnecessary for
other transactions. That means the changes may be either cleaned up by
vacuum or removed from HOT chains (thus made inaccessible through
indexes), and there may be other such consequences.

When decoding committed transactions this is not an issue, and we
never decode transactions that abort before the decoding starts.

But for in-progress transactions, this may cause failures when the
output plugin consults catalogs (both system and user-defined).

We handle such failures by returning ERRCODE_TRANSACTION_ROLLBACK
sqlerrcode from system table scan APIs to the backend decoding a
specific uncommitted transaction. The decoding logic on the receipt
of such an sqlerrcode aborts the ongoing decoding and returns
gracefully.
---
 doc/src/sgml/logicaldecoding.sgml  | 10 ++++---
 src/backend/access/heap/heapam.c   |  8 ++++++
 src/backend/access/index/genam.c   | 53 ++++++++++++++++++++++++++++++++++++++
 src/backend/access/table/tableam.c |  8 ++++++
 src/backend/utils/time/snapmgr.c   | 13 +++++++++-
 src/include/access/tableam.h       | 41 +++++++++++++++++++++++++++++
 src/include/utils/snapmgr.h        |  2 ++
 7 files changed, 131 insertions(+), 4 deletions(-)

diff --git a/doc/src/sgml/logicaldecoding.sgml b/doc/src/sgml/logicaldecoding.sgml
index 65244b1..979844c 100644
--- a/doc/src/sgml/logicaldecoding.sgml
+++ b/doc/src/sgml/logicaldecoding.sgml
@@ -432,9 +432,13 @@ typedef void (*LogicalOutputPluginInit) (struct OutputPluginCallbacks *cb);
 ALTER TABLE user_catalog_table SET (user_catalog_table = true);
 CREATE TABLE another_catalog_table(data text) WITH (user_catalog_table = true);
 </programlisting>
-     Any actions leading to transaction ID assignment are prohibited. That, among others,
-     includes writing to tables, performing DDL changes, and
-     calling <literal>pg_current_xact_id()</literal>.
+     Note that access to user catalog tables or regular system catalog tables in
+     the output plugins has to be done via the <literal>systable_*</literal> scan
+     APIs only. The user tables should not be accesed in the output plugins anyways.
+     Access via the <literal>heap_*</literal> scan APIs will error out. Additionally,
+     any actions leading to transaction ID assignment are prohibited. That, among
+     others, includes writing to tables, performing DDL changes, and calling
+     <literal>pg_current_xact_id()</literal>.
     </para>
    </sect2>
 
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 0d4ed60..d854c45 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -1288,6 +1288,14 @@ heap_getnext(TableScanDesc sscan, ScanDirection direction)
 				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 				 errmsg_internal("only heap AM is supported")));
 
+	/*
+	 * We don't expect direct calls to this routine when CheckXidAlive is a
+	 * valid transaction id, this should only come through systable_* call.
+	 * CheckXidAlive is set during logical decoding of a transactions.
+	 */
+	if (unlikely(TransactionIdIsValid(CheckXidAlive) && !sysbegin_called))
+		elog(ERROR, "unexpected heap_getnext call during logical decoding");
+
 	/* Note: no locking manipulations needed */
 
 	if (scan->rs_base.rs_flags & SO_ALLOW_PAGEMODE)
diff --git a/src/backend/access/index/genam.c b/src/backend/access/index/genam.c
index dfba5ae..77cedd9 100644
--- a/src/backend/access/index/genam.c
+++ b/src/backend/access/index/genam.c
@@ -28,6 +28,7 @@
 #include "lib/stringinfo.h"
 #include "miscadmin.h"
 #include "storage/bufmgr.h"
+#include "storage/procarray.h"
 #include "utils/acl.h"
 #include "utils/builtins.h"
 #include "utils/lsyscache.h"
@@ -429,10 +430,37 @@ systable_beginscan(Relation heapRelation,
 		sysscan->iscan = NULL;
 	}
 
+	/*
+	 * If CheckXidAlive is set then set a flag that this call is passed through
+	 * systable_beginscan.  See detailed  comments at snapmgr.c where these
+	 * variables are declared.
+	 */
+	if (TransactionIdIsValid(CheckXidAlive))
+		sysbegin_called = true;
+
 	return sysscan;
 }
 
 /*
+ * HandleConcurrentAbort - Handle concurrent abort of the CheckXidAlive.
+ *
+ * If CheckXidAlive is valid, then we check if it aborted. If it did, we error
+ * out.  We can't directly use TransactionIdDidAbort as after crash such
+ * transaction might not have been marked as aborted.  See detailed  comments
+ * at snapmgr.c where the variable is declared.
+ */
+static inline void
+HandleConcurrentAbort()
+{
+	if (TransactionIdIsValid(CheckXidAlive) &&
+			!TransactionIdIsInProgress(CheckXidAlive) &&
+			!TransactionIdDidCommit(CheckXidAlive))
+			ereport(ERROR,
+				(errcode(ERRCODE_TRANSACTION_ROLLBACK),
+				 errmsg("transaction aborted during system catalog scan")));
+}
+
+/*
  * systable_getnext --- get next tuple in a heap-or-index scan
  *
  * Returns NULL if no more tuples available.
@@ -481,6 +509,12 @@ systable_getnext(SysScanDesc sysscan)
 		}
 	}
 
+	/*
+	 * Handle the concurrent abort while fetching the catalog tuple during
+	 * logical streaming of a transaction.
+	 */
+	HandleConcurrentAbort();
+
 	return htup;
 }
 
@@ -517,6 +551,12 @@ systable_recheck_tuple(SysScanDesc sysscan, HeapTuple tup)
 											sysscan->slot,
 											freshsnap);
 
+	/*
+	 * Handle the concurrent abort while fetching the catalog tuple during
+	 * logical streaming of a transaction.
+	 */
+	HandleConcurrentAbort();
+
 	return result;
 }
 
@@ -545,6 +585,13 @@ systable_endscan(SysScanDesc sysscan)
 	if (sysscan->snapshot)
 		UnregisterSnapshot(sysscan->snapshot);
 
+	/*
+	 * Reset the sysbegin_called flag at the end of the systable scan.  See
+	 * detailed  comments at snapmgr.c where these variables are declared.
+	 */
+	if (TransactionIdIsValid(CheckXidAlive))
+		sysbegin_called = false;
+
 	pfree(sysscan);
 }
 
@@ -643,6 +690,12 @@ systable_getnext_ordered(SysScanDesc sysscan, ScanDirection direction)
 	if (htup && sysscan->iscan->xs_recheck)
 		elog(ERROR, "system catalog scans with lossy index conditions are not implemented");
 
+	/*
+	 * Handle the concurrent abort while fetching the catalog tuple during
+	 * logical streaming of a transaction.
+	 */
+	HandleConcurrentAbort();
+
 	return htup;
 }
 
diff --git a/src/backend/access/table/tableam.c b/src/backend/access/table/tableam.c
index c814733..e02faa5 100644
--- a/src/backend/access/table/tableam.c
+++ b/src/backend/access/table/tableam.c
@@ -231,6 +231,14 @@ table_tuple_get_latest_tid(TableScanDesc scan, ItemPointer tid)
 	const TableAmRoutine *tableam = rel->rd_tableam;
 
 	/*
+	 * We don't expect direct calls to this routine when CheckXidAlive is a
+	 * valid transaction id, this should only come through systable_* call.
+	 * CheckXidAlive is set during logical decoding of a transactions.
+	 */
+	if (unlikely(TransactionIdIsValid(CheckXidAlive) && !sysbegin_called))
+		elog(ERROR, "unexpected table_tuple_get_latest_tid call during logical decoding");
+
+	/*
 	 * Since this can be called with user-supplied TID, don't trust the input
 	 * too much.
 	 */
diff --git a/src/backend/utils/time/snapmgr.c b/src/backend/utils/time/snapmgr.c
index 1c063c5..40af75c 100644
--- a/src/backend/utils/time/snapmgr.c
+++ b/src/backend/utils/time/snapmgr.c
@@ -154,6 +154,18 @@ static Snapshot CatalogSnapshot = NULL;
 static Snapshot HistoricSnapshot = NULL;
 
 /*
+ * An xid value pointing to a possibly ongoing (sub)transaction.
+ * Currently used in logical decoding.  It's possible that such transactions
+ * can get aborted while the decoding is ongoing.  If CheckXidAlive is set
+ * then we will set sysbegin_called flag when we call systable_beginscan.  This
+ * is to ensure that from the pgoutput plugin we should never directly access
+ * the tableam or heap apis because we are checking for the concurrent abort
+ * only in systable_* apis.
+ */
+TransactionId CheckXidAlive = InvalidTransactionId;
+bool sysbegin_called = false;
+
+/*
  * These are updated by GetSnapshotData.  We initialize them this way
  * for the convenience of TransactionIdIsInProgress: even in bootstrap
  * mode, we don't want it to say that BootstrapTransactionId is in progress.
@@ -2043,7 +2055,6 @@ SetupHistoricSnapshot(Snapshot historic_snapshot, HTAB *tuplecids)
 	tuplecid_data = tuplecids;
 }
 
-
 /*
  * Make catalog snapshots behave normally again.
  */
diff --git a/src/include/access/tableam.h b/src/include/access/tableam.h
index 94903dd..4daff77 100644
--- a/src/include/access/tableam.h
+++ b/src/include/access/tableam.h
@@ -21,6 +21,7 @@
 #include "access/sdir.h"
 #include "utils/guc.h"
 #include "utils/rel.h"
+#include "utils/snapmgr.h"
 #include "utils/snapshot.h"
 
 
@@ -903,6 +904,15 @@ static inline bool
 table_scan_getnextslot(TableScanDesc sscan, ScanDirection direction, TupleTableSlot *slot)
 {
 	slot->tts_tableOid = RelationGetRelid(sscan->rs_rd);
+
+	/*
+	 * We don't expect direct calls to this routine when CheckXidAlive is a
+	 * valid transaction id.  CheckXidAlive is set during logical decoding of
+	 * a transactions.
+	 */
+	if (unlikely(TransactionIdIsValid(CheckXidAlive) && !sysbegin_called))
+		elog(ERROR, "unexpected table_scan_getnextslot call during logical decoding");
+
 	return sscan->rs_rd->rd_tableam->scan_getnextslot(sscan, direction, slot);
 }
 
@@ -1015,6 +1025,13 @@ table_index_fetch_tuple(struct IndexFetchTableData *scan,
 						TupleTableSlot *slot,
 						bool *call_again, bool *all_dead)
 {
+	/*
+	 * We don't expect direct calls to this routine when CheckXidAlive is a
+	 * valid transaction id, this should only come through systable_* call.
+	 * CheckXidAlive is set during logical decoding of a transactions.
+	 */
+	if (unlikely(TransactionIdIsValid(CheckXidAlive) && !sysbegin_called))
+		elog(ERROR, "unexpected table_index_fetch_tuple call during logical decoding");
 
 	return scan->rel->rd_tableam->index_fetch_tuple(scan, tid, snapshot,
 													slot, call_again,
@@ -1054,6 +1071,14 @@ table_tuple_fetch_row_version(Relation rel,
 							  Snapshot snapshot,
 							  TupleTableSlot *slot)
 {
+	/*
+	 * We don't expect direct calls to this routine when CheckXidAlive is a
+	 * valid transaction id, this should only come through systable_* call.
+	 * CheckXidAlive is set during logical decoding of a transactions.
+	 */
+	if (unlikely(TransactionIdIsValid(CheckXidAlive) && !sysbegin_called))
+		elog(ERROR, "unexpected table_tuple_fetch_row_version call during logical decoding");
+
 	return rel->rd_tableam->tuple_fetch_row_version(rel, tid, snapshot, slot);
 }
 
@@ -1710,6 +1735,14 @@ static inline bool
 table_scan_bitmap_next_block(TableScanDesc scan,
 							 struct TBMIterateResult *tbmres)
 {
+	/*
+	 * We don't expect direct calls to this routine when CheckXidAlive is a
+	 * valid transaction id, this should only come through systable_* call.
+	 * CheckXidAlive is set during logical decoding of a transactions.
+	 */
+	if (unlikely(TransactionIdIsValid(CheckXidAlive) && !sysbegin_called))
+		elog(ERROR, "unexpected table_scan_bitmap_next_block call during logical decoding");
+
 	return scan->rs_rd->rd_tableam->scan_bitmap_next_block(scan,
 														   tbmres);
 }
@@ -1727,6 +1760,14 @@ table_scan_bitmap_next_tuple(TableScanDesc scan,
 							 struct TBMIterateResult *tbmres,
 							 TupleTableSlot *slot)
 {
+	/*
+	 * We don't expect direct calls to this routine when CheckXidAlive is a
+	 * valid transaction id, this should only come through systable_* call.
+	 * CheckXidAlive is set during logical decoding of a transactions.
+	 */
+	if (unlikely(TransactionIdIsValid(CheckXidAlive) && !sysbegin_called))
+		elog(ERROR, "unexpected table_scan_bitmap_next_tuple call during logical decoding");
+
 	return scan->rs_rd->rd_tableam->scan_bitmap_next_tuple(scan,
 														   tbmres,
 														   slot);
diff --git a/src/include/utils/snapmgr.h b/src/include/utils/snapmgr.h
index b28d13c..ae1cbe4 100644
--- a/src/include/utils/snapmgr.h
+++ b/src/include/utils/snapmgr.h
@@ -145,6 +145,8 @@ extern bool XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot);
 
 /* Support for catalog timetravel for logical decoding */
 struct HTAB;
+extern TransactionId CheckXidAlive;
+extern bool	sysbegin_called;
 extern struct HTAB *HistoricSnapshotGetTupleCids(void);
 extern void SetupHistoricSnapshot(Snapshot snapshot_now, struct HTAB *tuplecids);
 extern void TeardownHistoricSnapshot(bool is_error);
-- 
1.8.3.1