4 files changed, 278 insertions, 23 deletions

diff --git a/src/backend/utils/activity/pgstat_function.c b/src/backend/utils/activity/pgstat_function.c
index 93ec29757aa..ad9879afb2a 100644
--- a/src/backend/utils/activity/pgstat_function.c
+++ b/src/backend/utils/activity/pgstat_function.c
@@ -49,6 +49,28 @@ static instr_time total_func_time;
 
 
 /*
+ * Ensure that stats are dropped if transaction aborts.
+ */
+void
+pgstat_create_function(Oid proid)
+{
+	pgstat_create_transactional(PGSTAT_KIND_FUNCTION,
+								MyDatabaseId,
+								proid);
+}
+
+/*
+ * Ensure that stats are dropped if transaction commits.
+ */
+void
+pgstat_drop_function(Oid proid)
+{
+	pgstat_drop_transactional(PGSTAT_KIND_FUNCTION,
+							  MyDatabaseId,
+							  proid);
+}
+
+/*
  * Initialize function call usage data.
  * Called by the executor before invoking a function.
  */
diff --git a/src/backend/utils/activity/pgstat_relation.c b/src/backend/utils/activity/pgstat_relation.c
index 4f97d2f1d9c..5b9b6dd7c60 100644
--- a/src/backend/utils/activity/pgstat_relation.c
+++ b/src/backend/utils/activity/pgstat_relation.c
@@ -171,33 +171,26 @@ pgstat_relation_init(Relation rel)
 }
 
 /*
- * Tell the collector that we just dropped a relation.
- * (If the message gets lost, we will still clean the dead entry eventually
- * via future invocations of pgstat_vacuum_stat().)
- *
- * Currently not used for lack of any good place to call it; we rely
- * entirely on pgstat_vacuum_stat() to clean out stats for dead rels.
+ * Ensure that stats are dropped if transaction aborts.
  */
-#ifdef NOT_USED
 void
-pgstat_drop_relation(Oid relid)
+pgstat_create_relation(Relation rel)
 {
-	PgStat_MsgTabpurge msg;
-	int			len;
-
-	if (pgStatSock == PGINVALID_SOCKET)
-		return;
-
-	msg.m_tableid[0] = relid;
-	msg.m_nentries = 1;
-
-	len = offsetof(PgStat_MsgTabpurge, m_tableid[0]) + sizeof(Oid);
+	pgstat_create_transactional(PGSTAT_KIND_RELATION,
+								rel->rd_rel->relisshared ? InvalidOid : MyDatabaseId,
+								RelationGetRelid(rel));
+}
 
-	pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_TABPURGE);
-	msg.m_databaseid = MyDatabaseId;
-	pgstat_send(&msg, len);
+/*
+ * Ensure that stats are dropped if transaction commits.
+ */
+void
+pgstat_drop_relation(Relation rel)
+{
+	pgstat_drop_transactional(PGSTAT_KIND_RELATION,
+							  rel->rd_rel->relisshared ? InvalidOid : MyDatabaseId,
+							  RelationGetRelid(rel));
 }
-#endif							/* NOT_USED */
 
 /*
  * Report that the table was just vacuumed.
diff --git a/src/backend/utils/activity/pgstat_subscription.c b/src/backend/utils/activity/pgstat_subscription.c
index 503dcabd204..689029b30af 100644
--- a/src/backend/utils/activity/pgstat_subscription.c
+++ b/src/backend/utils/activity/pgstat_subscription.c
@@ -35,14 +35,31 @@ pgstat_report_subscription_error(Oid subid, bool is_apply_error)
 }
 
 /*
+ * Report creating the subscription.
+ *
+ * Ensures that stats are dropped if transaction rolls back.
+ */
+void
+pgstat_create_subscription(Oid subid)
+{
+	pgstat_create_transactional(PGSTAT_KIND_SUBSCRIPTION,
+								InvalidOid, subid);
+}
+
+/*
  * Report dropping the subscription.
+ *
+ * Ensures that stats are dropped if transaction commits.
  */
 void
-pgstat_report_subscription_drop(Oid subid)
+pgstat_drop_subscription(Oid subid)
 {
 	PgStat_MsgSubscriptionDrop msg;
 
 	pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_SUBSCRIPTIONDROP);
 	msg.m_subid = subid;
 	pgstat_send(&msg, sizeof(PgStat_MsgSubscriptionDrop));
+
+	pgstat_drop_transactional(PGSTAT_KIND_SUBSCRIPTION,
+							  InvalidOid, subid);
 }
diff --git a/src/backend/utils/activity/pgstat_xact.c b/src/backend/utils/activity/pgstat_xact.c
index 17907e32789..5c00eab7c70 100644
--- a/src/backend/utils/activity/pgstat_xact.c
+++ b/src/backend/utils/activity/pgstat_xact.c
@@ -19,6 +19,18 @@
 #include "utils/pgstat_internal.h"
 
 
+typedef struct PgStat_PendingDroppedStatsItem
+{
+	xl_xact_stats_item item;
+	bool		is_create;
+	dlist_node	node;
+} PgStat_PendingDroppedStatsItem;
+
+
+static void AtEOXact_PgStat_DroppedStats(PgStat_SubXactStatus *xact_state, bool isCommit);
+static void AtEOSubXact_PgStat_DroppedStats(PgStat_SubXactStatus *xact_state,
+											bool isCommit, int nestDepth);
+
 static PgStat_SubXactStatus *pgStatXactStack = NULL;
 
 
@@ -40,6 +52,7 @@ AtEOXact_PgStat(bool isCommit, bool parallel)
 		Assert(xact_state->prev == NULL);
 
 		AtEOXact_PgStat_Relations(xact_state, isCommit);
+		AtEOXact_PgStat_DroppedStats(xact_state, isCommit);
 	}
 	pgStatXactStack = NULL;
 
@@ -48,6 +61,49 @@ AtEOXact_PgStat(bool isCommit, bool parallel)
 }
 
 /*
+ * When committing, drop stats for objects dropped in the transaction. When
+ * aborting, drop stats for objects created in the transaction.
+ */
+static void
+AtEOXact_PgStat_DroppedStats(PgStat_SubXactStatus *xact_state, bool isCommit)
+{
+	dlist_mutable_iter iter;
+
+	if (xact_state->pending_drops_count == 0)
+	{
+		Assert(dlist_is_empty(&xact_state->pending_drops));
+		return;
+	}
+
+	dlist_foreach_modify(iter, &xact_state->pending_drops)
+	{
+		PgStat_PendingDroppedStatsItem *pending =
+		dlist_container(PgStat_PendingDroppedStatsItem, node, iter.cur);
+
+		if (isCommit && !pending->is_create)
+		{
+			/*
+			 * Transaction that dropped an object committed. Drop the stats
+			 * too.
+			 */
+			/* will do work in subsequent commit */
+		}
+		else if (!isCommit && pending->is_create)
+		{
+			/*
+			 * Transaction that created an object aborted. Drop the stats
+			 * associated with the object.
+			 */
+			/* will do work in subsequent commit */
+		}
+
+		dlist_delete(&pending->node);
+		xact_state->pending_drops_count--;
+		pfree(pending);
+	}
+}
+
+/*
  * Called from access/transam/xact.c at subtransaction commit/abort.
  */
 void
@@ -64,12 +120,64 @@ AtEOSubXact_PgStat(bool isCommit, int nestDepth)
 		pgStatXactStack = xact_state->prev;
 
 		AtEOSubXact_PgStat_Relations(xact_state, isCommit, nestDepth);
+		AtEOSubXact_PgStat_DroppedStats(xact_state, isCommit, nestDepth);
 
 		pfree(xact_state);
 	}
 }
 
 /*
+ * Like AtEOXact_PgStat_DroppedStats(), but for subtransactions.
+ */
+static void
+AtEOSubXact_PgStat_DroppedStats(PgStat_SubXactStatus *xact_state,
+								bool isCommit, int nestDepth)
+{
+	PgStat_SubXactStatus *parent_xact_state;
+	dlist_mutable_iter iter;
+
+	if (xact_state->pending_drops_count == 0)
+		return;
+
+	parent_xact_state = pgstat_xact_stack_level_get(nestDepth - 1);
+
+	dlist_foreach_modify(iter, &xact_state->pending_drops)
+	{
+		PgStat_PendingDroppedStatsItem *pending =
+		dlist_container(PgStat_PendingDroppedStatsItem, node, iter.cur);
+
+		dlist_delete(&pending->node);
+		xact_state->pending_drops_count--;
+
+		if (!isCommit && pending->is_create)
+		{
+			/*
+			 * Subtransaction creating a new stats object aborted. Drop the
+			 * stats object.
+			 */
+			/* will do work in subsequent commit */
+			pfree(pending);
+		}
+		else if (isCommit)
+		{
+			/*
+			 * Subtransaction dropping a stats object committed. Can't yet
+			 * remove the stats object, the surrounding transaction might
+			 * still abort. Pass it on to the parent.
+			 */
+			dlist_push_tail(&parent_xact_state->pending_drops, &pending->node);
+			parent_xact_state->pending_drops_count++;
+		}
+		else
+		{
+			pfree(pending);
+		}
+	}
+
+	Assert(xact_state->pending_drops_count == 0);
+}
+
+/*
  * Save the transactional stats state at 2PC transaction prepare.
  */
 void
@@ -130,6 +238,8 @@ pgstat_xact_stack_level_get(int nest_level)
 		xact_state = (PgStat_SubXactStatus *)
 			MemoryContextAlloc(TopTransactionContext,
 							   sizeof(PgStat_SubXactStatus));
+		dlist_init(&xact_state->pending_drops);
+		xact_state->pending_drops_count = 0;
 		xact_state->nest_level = nest_level;
 		xact_state->prev = pgStatXactStack;
 		xact_state->first = NULL;
@@ -137,3 +247,116 @@ pgstat_xact_stack_level_get(int nest_level)
 	}
 	return xact_state;
 }
+
+/*
+ * Get stat items that need to be dropped at commit / abort.
+ *
+ * When committing, stats for objects that have been dropped in the
+ * transaction are returned. When aborting, stats for newly created objects are
+ * returned.
+ *
+ * Used by COMMIT / ABORT and 2PC PREPARE processing when building their
+ * respective WAL records, to ensure stats are dropped in case of a crash / on
+ * standbys.
+ *
+ * The list of items is allocated in CurrentMemoryContext and must be freed by
+ * the caller (directly or via memory context reset).
+ */
+int
+pgstat_get_transactional_drops(bool isCommit, xl_xact_stats_item **items)
+{
+	PgStat_SubXactStatus *xact_state = pgStatXactStack;
+	int			nitems = 0;
+	dlist_iter	iter;
+
+	if (xact_state == NULL)
+		return 0;
+
+	/*
+	 * We expect to be called for subtransaction abort (which logs a WAL
+	 * record), but not for subtransaction commit (which doesn't).
+	 */
+	Assert(!isCommit || xact_state->nest_level == 1);
+	Assert(!isCommit || xact_state->prev == NULL);
+
+	*items = palloc(xact_state->pending_drops_count
+					* sizeof(xl_xact_stats_item));
+
+	dlist_foreach(iter, &xact_state->pending_drops)
+	{
+		PgStat_PendingDroppedStatsItem *pending =
+		dlist_container(PgStat_PendingDroppedStatsItem, node, iter.cur);
+
+		if (isCommit && pending->is_create)
+			continue;
+		if (!isCommit && !pending->is_create)
+			continue;
+
+		Assert(nitems < xact_state->pending_drops_count);
+		(*items)[nitems++] = pending->item;
+	}
+
+	return nitems;
+}
+
+/*
+ * Execute scheduled drops post-commit. Called from xact_redo_commit() /
+ * xact_redo_abort() during recovery, and from FinishPreparedTransaction()
+ * during normal 2PC COMMIT/ABORT PREPARED processing.
+ */
+void
+pgstat_execute_transactional_drops(int ndrops, struct xl_xact_stats_item *items, bool is_redo)
+{
+	if (ndrops == 0)
+		return;
+
+	for (int i = 0; i < ndrops; i++)
+	{
+		/* will do work in subsequent commit */
+	}
+}
+
+static void
+create_drop_transactional_internal(PgStat_Kind kind, Oid dboid, Oid objoid, bool is_create)
+{
+	int			nest_level = GetCurrentTransactionNestLevel();
+	PgStat_SubXactStatus *xact_state;
+	PgStat_PendingDroppedStatsItem *drop = (PgStat_PendingDroppedStatsItem *)
+	MemoryContextAlloc(TopTransactionContext, sizeof(PgStat_PendingDroppedStatsItem));
+
+	xact_state = pgstat_xact_stack_level_get(nest_level);
+
+	drop->is_create = is_create;
+	drop->item.kind = kind;
+	drop->item.dboid = dboid;
+	drop->item.objoid = objoid;
+
+	dlist_push_tail(&xact_state->pending_drops, &drop->node);
+	xact_state->pending_drops_count++;
+}
+
+/*
+ * Create a stats entry for a newly created database object in a transactional
+ * manner.
+ *
+ * I.e. if the current (sub-)transaction aborts, the stats entry will also be
+ * dropped.
+ */
+void
+pgstat_create_transactional(PgStat_Kind kind, Oid dboid, Oid objoid)
+{
+	create_drop_transactional_internal(kind, dboid, objoid, /* create */ true);
+}
+
+/*
+ * Drop a stats entry for a just dropped database object in a transactional
+ * manner.
+ *
+ * I.e. if the current (sub-)transaction aborts, the stats entry will stay
+ * alive.
+ */
+void
+pgstat_drop_transactional(PgStat_Kind kind, Oid dboid, Oid objoid)
+{
+	create_drop_transactional_internal(kind, dboid, objoid, /* create */ false);
+}