Optimizing ResouceOwner to speed up COPY

From bbb37408e5496fd011cf460279cf9ba2533c4fd6 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <tomas@vondra.me>
Date: Thu, 16 Oct 2025 13:27:25 +0200
Subject: [PATCH v2] Deduplicate entries in ResourceOwner

We may track the same resource many times, but the current ResourceOwner
does not handle that efficiently. Each reference is a separate entry,
which leads to long linear searches in the simple hash table.

This adds a simple opportunistic deduplication. Entries get a counter,
tracking how many references they represent. It only applies to the
"hash" phase, not to the initial array. And when adding an entry to the
hash finds an empty slot first, it uses it (instead of continuing to
look for a possible match).

This means the deduplication is not perfect - there may be multiple
entries for the same value. But in practice it seems like a good trade
off. It won't affect cases with no duplicates, and it will help cases
with many of them.

The code also continues to grow the hash table as before. In some cases
it might be enough to deduplicate the contents. But it's hard to say in
advance, and it's not much cheaper as it still requires walking the
current hash table. So there's a risk of attempting deduplication, only
to find the hash still needs to be enlarged, doubling the cost. So we
just grow the hash table. In the worst case the memory usage is not
worse than without deduplication. If deduplication is effective, the
hash table grows later or not at all.
---
 src/backend/utils/resowner/resowner.c | 215 ++++++++++++++++++++------
 1 file changed, 166 insertions(+), 49 deletions(-)

diff --git a/src/backend/utils/resowner/resowner.c b/src/backend/utils/resowner/resowner.c
index fca84ded6dd..055426c7761 100644
--- a/src/backend/utils/resowner/resowner.c
+++ b/src/backend/utils/resowner/resowner.c
@@ -15,6 +15,12 @@
  * a particular reference, you need to search both the array and the hash
  * table.
  *
+ * The hash table deduplicates entries in an opportunistic way. If it finds
+ * an entry for the same resource (same kind), it increments a counter instead.
+ * But if it finds an empty slot first, it uses the slot. If the table gets
+ * enlarged, those entries will be merged, but it's possible to have multiple
+ * entries for the same resource. The fixed-size array does not deduplicate.
+ *
  * The most frequent usage is that a resource is remembered, and forgotten
  * shortly thereafter.  For example, pin a buffer, read one tuple from it,
  * release the pin.  Linearly scanning the small array handles that case
@@ -67,6 +73,18 @@ typedef struct ResourceElem
 	const ResourceOwnerDesc *kind;	/* NULL indicates a free hash table slot */
 } ResourceElem;
 
+/*
+ * ResourceHashElem represents a reference associated with a resource owner,
+ * stored in the hash table. Similar to ResourceElem, but has a counter field
+ * tracking how many references it represents, to allow deduplication.
+ */
+typedef struct ResourceHashElem
+{
+ 	Datum		item;
+	uint32		count;				/* number of references */
+	const ResourceOwnerDesc *kind;	/* NULL indicates a free hash table slot */
+} ResourceHashElem;
+
 /*
  * Size of the fixed-size array to hold most-recently remembered resources.
  */
@@ -151,7 +169,7 @@ struct ResourceOwnerData
 	 * release priority.  The first 'nhash' elements are occupied, the rest
 	 * are empty.
 	 */
-	ResourceElem *hash;
+	ResourceHashElem *hash;
 	uint32		capacity;		/* allocated length of hash[] */
 	uint32		grow_at;		/* grow hash when reach this */
 
@@ -198,7 +216,8 @@ static ResourceReleaseCallbackItem *ResourceRelease_callbacks = NULL;
 /* Internal routines */
 static inline uint32 hash_resource_elem(Datum value, const ResourceOwnerDesc *kind);
 static void ResourceOwnerAddToHash(ResourceOwner owner, Datum value,
-								   const ResourceOwnerDesc *kind);
+								   const ResourceOwnerDesc *kind,
+								   uint32 count);
 static int	resource_priority_cmp(const void *a, const void *b);
 static void ResourceOwnerSort(ResourceOwner owner);
 static void ResourceOwnerReleaseAll(ResourceOwner owner,
@@ -239,7 +258,8 @@ hash_resource_elem(Datum value, const ResourceOwnerDesc *kind)
  * Adds 'value' of given 'kind' to the ResourceOwner's hash table
  */
 static void
-ResourceOwnerAddToHash(ResourceOwner owner, Datum value, const ResourceOwnerDesc *kind)
+ResourceOwnerAddToHash(ResourceOwner owner, Datum value, const ResourceOwnerDesc *kind,
+					   uint32 count)
 {
 	uint32		mask = owner->capacity - 1;
 	uint32		idx;
@@ -250,12 +270,25 @@ ResourceOwnerAddToHash(ResourceOwner owner, Datum value, const ResourceOwnerDesc
 	idx = hash_resource_elem(value, kind) & mask;
 	for (;;)
 	{
+		/*
+		 * Increment counter for a matching entry or use an empty slot,
+		 * whichever we find first.
+		 */
+
+		if ((owner->hash[idx].kind == kind) &&
+			(owner->hash[idx].item == value))
+		{
+			owner->hash[idx].count += count;
+			return;
+		}
+
 		if (owner->hash[idx].kind == NULL)
 			break;				/* found a free slot */
 		idx = (idx + 1) & mask;
 	}
 	owner->hash[idx].item = value;
 	owner->hash[idx].kind = kind;
+	owner->hash[idx].count = count;
 	owner->nhash++;
 }
 
@@ -277,6 +310,24 @@ resource_priority_cmp(const void *a, const void *b)
 		return 1;
 }
 
+/*
+ * Comparison function to sort by release phase and priority
+ */
+static int
+resource_priority_hash_cmp(const void *a, const void *b)
+{
+	const ResourceHashElem *ra = (const ResourceHashElem *) a;
+	const ResourceHashElem *rb = (const ResourceHashElem *) b;
+
+	/* Note: reverse order */
+	if (ra->kind->release_phase == rb->kind->release_phase)
+		return pg_cmp_u32(rb->kind->release_priority, ra->kind->release_priority);
+	else if (ra->kind->release_phase > rb->kind->release_phase)
+		return -1;
+	else
+		return 1;
+}
+
 /*
  * Sort resources in reverse release priority.
  *
@@ -288,16 +339,21 @@ resource_priority_cmp(const void *a, const void *b)
 static void
 ResourceOwnerSort(ResourceOwner owner)
 {
-	ResourceElem *items;
-	uint32		nitems;
-
 	if (owner->nhash == 0)
 	{
+		ResourceElem *items;
+		uint32		nitems;
+
 		items = owner->arr;
 		nitems = owner->narr;
+
+		qsort(items, nitems, sizeof(ResourceElem), resource_priority_cmp);
 	}
 	else
 	{
+		ResourceHashElem   *items;
+		uint32				nitems;
+
 		/*
 		 * Compact the hash table, so that all the elements are in the
 		 * beginning of the 'hash' array, with no empty elements.
@@ -324,7 +380,9 @@ ResourceOwnerSort(ResourceOwner owner)
 		Assert(dst + owner->narr <= owner->capacity);
 		for (int idx = 0; idx < owner->narr; idx++)
 		{
-			owner->hash[dst] = owner->arr[idx];
+			owner->hash[dst].item = owner->arr[idx].item;
+			owner->hash[dst].kind = owner->arr[idx].kind;
+			owner->hash[dst].count = 1;
 			dst++;
 		}
 		Assert(dst == owner->nhash + owner->narr);
@@ -333,9 +391,9 @@ ResourceOwnerSort(ResourceOwner owner)
 
 		items = owner->hash;
 		nitems = owner->nhash;
-	}
 
-	qsort(items, nitems, sizeof(ResourceElem), resource_priority_cmp);
+		qsort(items, nitems, sizeof(ResourceHashElem), resource_priority_hash_cmp);
+	}
 }
 
 /*
@@ -345,9 +403,6 @@ static void
 ResourceOwnerReleaseAll(ResourceOwner owner, ResourceReleasePhase phase,
 						bool printLeakWarnings)
 {
-	ResourceElem *items;
-	uint32		nitems;
-
 	/*
 	 * ResourceOwnerSort must've been called already.  All the resources are
 	 * either in the array or the hash.
@@ -356,49 +411,93 @@ ResourceOwnerReleaseAll(ResourceOwner owner, ResourceReleasePhase phase,
 	Assert(owner->sorted);
 	if (owner->nhash == 0)
 	{
+		ResourceElem *items;
+		uint32		nitems;
+
 		items = owner->arr;
 		nitems = owner->narr;
+
+		/*
+		 * The resources are sorted in reverse priority order.  Release them
+		 * starting from the end, until we hit the end of the phase that we are
+		 * releasing now.  We will continue from there when called again for the
+		 * next phase.
+		 */
+		while (nitems > 0)
+		{
+			uint32		idx = nitems - 1;
+			Datum		value = items[idx].item;
+			const ResourceOwnerDesc *kind = items[idx].kind;
+
+			if (kind->release_phase > phase)
+				break;
+			Assert(kind->release_phase == phase);
+
+			if (printLeakWarnings)
+			{
+				char	   *res_str;
+
+				res_str = kind->DebugPrint ?
+					kind->DebugPrint(value)
+					: psprintf("%s %p", kind->name, DatumGetPointer(value));
+				pfree(res_str);
+			}
+
+			kind->ReleaseResource(value);
+
+			nitems--;
+		}
+
+		owner->narr = nitems;
 	}
 	else
 	{
+		ResourceHashElem *items;
+		uint32		nitems;
+
 		Assert(owner->narr == 0);
 		items = owner->hash;
 		nitems = owner->nhash;
-	}
 
-	/*
-	 * The resources are sorted in reverse priority order.  Release them
-	 * starting from the end, until we hit the end of the phase that we are
-	 * releasing now.  We will continue from there when called again for the
-	 * next phase.
-	 */
-	while (nitems > 0)
-	{
-		uint32		idx = nitems - 1;
-		Datum		value = items[idx].item;
-		const ResourceOwnerDesc *kind = items[idx].kind;
+		/*
+		 * The resources are sorted in reverse priority order.  Release them
+		 * starting from the end, until we hit the end of the phase that we are
+		 * releasing now.  We will continue from there when called again for the
+		 * next phase.
+		 */
+		while (nitems > 0)
+		{
+			uint32		idx = nitems - 1;
+			Datum		value = items[idx].item;
+			const ResourceOwnerDesc *kind = items[idx].kind;
+			uint32		count = items[idx].count;
 
-		if (kind->release_phase > phase)
-			break;
-		Assert(kind->release_phase == phase);
+			if (kind->release_phase > phase)
+				break;
+			Assert(kind->release_phase == phase);
 
-		if (printLeakWarnings)
-		{
-			char	   *res_str;
+			if (printLeakWarnings)
+			{
+				char	   *res_str;
+
+				res_str = kind->DebugPrint ?
+					kind->DebugPrint(value)
+					: psprintf("%s %p", kind->name, DatumGetPointer(value));
+				pfree(res_str);
+			}
+
+			/* invoke the release callback for each reference */
+			while (count > 0)
+			{
+				kind->ReleaseResource(value);
+				count--;
+			}
 
-			res_str = kind->DebugPrint ?
-				kind->DebugPrint(value)
-				: psprintf("%s %p", kind->name, DatumGetPointer(value));
-			elog(WARNING, "resource was not closed: %s", res_str);
-			pfree(res_str);
+			nitems--;
 		}
-		kind->ReleaseResource(value);
-		nitems--;
-	}
-	if (owner->nhash == 0)
-		owner->narr = nitems;
-	else
+
 		owner->nhash = nitems;
+	}
 }
 
 
@@ -466,16 +565,16 @@ ResourceOwnerEnlarge(ResourceOwner owner)
 		uint32		i,
 					oldcap,
 					newcap;
-		ResourceElem *oldhash;
-		ResourceElem *newhash;
+		ResourceHashElem *oldhash;
+		ResourceHashElem *newhash;
 
 		oldhash = owner->hash;
 		oldcap = owner->capacity;
 
 		/* Double the capacity (it must stay a power of 2!) */
 		newcap = (oldcap > 0) ? oldcap * 2 : RESOWNER_HASH_INIT_SIZE;
-		newhash = (ResourceElem *) MemoryContextAllocZero(TopMemoryContext,
-														  newcap * sizeof(ResourceElem));
+		newhash = (ResourceHashElem *) MemoryContextAllocZero(TopMemoryContext,
+														  newcap * sizeof(ResourceHashElem));
 
 		/*
 		 * We assume we can't fail below this point, so OK to scribble on the
@@ -496,7 +595,8 @@ ResourceOwnerEnlarge(ResourceOwner owner)
 			for (i = 0; i < oldcap; i++)
 			{
 				if (oldhash[i].kind != NULL)
-					ResourceOwnerAddToHash(owner, oldhash[i].item, oldhash[i].kind);
+					ResourceOwnerAddToHash(owner, oldhash[i].item, oldhash[i].kind,
+										   oldhash[i].count);
 			}
 
 			/* And release old hash table. */
@@ -506,7 +606,7 @@ ResourceOwnerEnlarge(ResourceOwner owner)
 
 	/* Move items from the array to the hash */
 	for (int i = 0; i < owner->narr; i++)
-		ResourceOwnerAddToHash(owner, owner->arr[i].item, owner->arr[i].kind);
+		ResourceOwnerAddToHash(owner, owner->arr[i].item, owner->arr[i].kind, 1);
 	owner->narr = 0;
 
 	Assert(owner->nhash <= owner->grow_at);
@@ -555,7 +655,8 @@ ResourceOwnerRemember(ResourceOwner owner, Datum value, const ResourceOwnerDesc
  * Note: Forgetting a resource does not guarantee that there is room to
  * remember a new resource.  One exception is when you forget the most
  * recently remembered resource; that does make room for a new remember call.
- * Some code callers rely on that exception.
+ * Some code callers rely on that exception. The deduplication does not
+ * change this, as it only applies to the hash table.
  */
 void
 ResourceOwnerForget(ResourceOwner owner, Datum value, const ResourceOwnerDesc *kind)
@@ -597,8 +698,17 @@ ResourceOwnerForget(ResourceOwner owner, Datum value, const ResourceOwnerDesc *k
 			if (owner->hash[idx].item == value &&
 				owner->hash[idx].kind == kind)
 			{
+				/* decrement the count for entry for multiple references */
+				if (owner->hash[idx].count > 1)
+				{
+					owner->hash[idx].count--;
+					return;
+				}
+
+				/* remove the entry entirely */
 				owner->hash[idx].item = (Datum) 0;
 				owner->hash[idx].kind = NULL;
+				owner->hash[idx].count = 0;
 				owner->nhash--;
 
 #ifdef RESOWNER_STATS
@@ -847,12 +957,19 @@ ResourceOwnerReleaseAllOfKind(ResourceOwner owner, const ResourceOwnerDesc *kind
 		if (owner->hash[i].kind == kind)
 		{
 			Datum		value = owner->hash[i].item;
+			uint32		count = owner->hash[i].count;
 
 			owner->hash[i].item = (Datum) 0;
 			owner->hash[i].kind = NULL;
+			owner->hash[i].count = 0;
 			owner->nhash--;
 
-			kind->ReleaseResource(value);
+			/* invoke the release callback once for each reference */
+			while (count > 0)
+			{
+				kind->ReleaseResource(value);
+				count--;
+			}
 		}
 	}
 	owner->releasing = false;
-- 
2.51.0

From c19d37c10cbe441fcd272bc67befca11d46fee21 Mon Sep 17 00:00:00 2001
From: Heikki Linnakangas <heikki.linnakangas@iki.fi>
Date: Tue, 21 Oct 2025 09:51:23 +0300
Subject: [PATCH v3 1/1] resownerbench

---
 contrib/Makefile                            |   1 +
 contrib/meson.build                         |   2 +
 contrib/resownerbench/Makefile              |  17 +++
 contrib/resownerbench/meson.build           |  24 +++
 contrib/resownerbench/resownerbench.c       | 154 ++++++++++++++++++++
 contrib/resownerbench/resownerbench.control |   6 +
 contrib/resownerbench/snaptest.sql          |  37 +++++
 7 files changed, 241 insertions(+)
 create mode 100644 contrib/resownerbench/Makefile
 create mode 100644 contrib/resownerbench/meson.build
 create mode 100644 contrib/resownerbench/resownerbench.c
 create mode 100644 contrib/resownerbench/resownerbench.control
 create mode 100644 contrib/resownerbench/snaptest.sql

diff --git a/contrib/Makefile b/contrib/Makefile
index 2f0a88d3f77..f84b26d52cf 100644
--- a/contrib/Makefile
+++ b/contrib/Makefile
@@ -43,6 +43,7 @@ SUBDIRS = \
 		pg_visibility	\
 		pg_walinspect	\
 		postgres_fdw	\
+		resownerbench	\
 		seg		\
 		spi		\
 		tablefunc	\
diff --git a/contrib/meson.build b/contrib/meson.build
index ed30ee7d639..9c34bb128b7 100644
--- a/contrib/meson.build
+++ b/contrib/meson.build
@@ -71,3 +71,5 @@ subdir('unaccent')
 subdir('uuid-ossp')
 subdir('vacuumlo')
 subdir('xml2')
+
+subdir('resownerbench')
diff --git a/contrib/resownerbench/Makefile b/contrib/resownerbench/Makefile
new file mode 100644
index 00000000000..9b0e1cfee1a
--- /dev/null
+++ b/contrib/resownerbench/Makefile
@@ -0,0 +1,17 @@
+MODULE_big = resownerbench
+OBJS = resownerbench.o
+
+EXTENSION = resownerbench
+DATA = resownerbench--1.0.sql
+PGFILEDESC = "resownerbench - benchmark for ResourceOwners"
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = contrib/resownerbench
+top_builddir = ../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
diff --git a/contrib/resownerbench/meson.build b/contrib/resownerbench/meson.build
new file mode 100644
index 00000000000..30051c3a375
--- /dev/null
+++ b/contrib/resownerbench/meson.build
@@ -0,0 +1,24 @@
+# Copyright (c) 2022-2025, PostgreSQL Global Development Group
+
+resownerbench_sources = files(
+  'resownerbench.c',
+)
+
+resownerbench = shared_module('resownerbench',
+  resownerbench_sources,
+  c_pch: pch_postgres_h,
+  kwargs: contrib_mod_args,
+)
+contrib_targets += resownerbench
+
+install_data(
+  'resownerbench.control',
+  'resownerbench--1.0.sql',
+  kwargs: contrib_data_args,
+)
+
+tests += {
+  'name': 'resownerbench',
+  'sd': meson.current_source_dir(),
+  'bd': meson.current_build_dir(),
+}
diff --git a/contrib/resownerbench/resownerbench.c b/contrib/resownerbench/resownerbench.c
new file mode 100644
index 00000000000..80f98c2105b
--- /dev/null
+++ b/contrib/resownerbench/resownerbench.c
@@ -0,0 +1,154 @@
+#include "postgres.h"
+
+#include "catalog/pg_type.h"
+#include "catalog/pg_statistic.h"
+#include "executor/spi.h"
+#include "funcapi.h"
+#include "libpq/pqsignal.h"
+#include "utils/catcache.h"
+#include "utils/syscache.h"
+#include "utils/timestamp.h"
+#include "utils/snapmgr.h"
+
+PG_MODULE_MAGIC;
+
+PG_FUNCTION_INFO_V1(snapshotbench_lifo);
+PG_FUNCTION_INFO_V1(snapshotbench_fifo);
+
+/*
+ * ResourceOwner Performance test, using RegisterSnapshot().
+ *
+ * This takes three parameters: numkeep, numsnaps, numiters.
+ *
+ * First, we register 'numkeep' snapshots. They are kept registed
+ * until the end of the test. Then, we repeatedly register and
+ * unregister 'numsnaps - numkeep' additional snapshots, repeating
+ * 'numiters' times. All the register/unregister calls are made in
+ * LIFO order.
+ *
+ * Returns the time spent, in milliseconds.
+ *
+ * The idea is to test the performance of ResourceOwnerRemember()
+ * and ReourceOwnerForget() operations, under different regimes.
+ *
+ * In the old implementation, if 'numsnaps' is small enough, all
+ * the entries fit in the resource owner's small array (it can
+ * hold 64 entries).
+ *
+ * In the new implementation, the array is much smaller, only 8
+ * entries, but it's used together with the hash table so that
+ * we stay in the "array regime" as long as 'numsnaps - numkeep'
+ * is smaller than 8 entries.
+ *
+ * 'numiters' can be adjusted to adjust the overall runtime to be
+ * suitable long.
+ */
+Datum
+snapshotbench_lifo(PG_FUNCTION_ARGS)
+{
+	int			numkeep = PG_GETARG_INT32(0);
+	int			numsnaps = PG_GETARG_INT32(1);
+	int			numiters = PG_GETARG_INT32(2);
+	int			i;
+	instr_time	start,
+				duration;
+	Snapshot	lsnap;
+	Snapshot   *rs;
+	int			numregistered = 0;
+
+	rs = palloc(Max(numsnaps, numkeep) * sizeof(Snapshot));
+
+	lsnap = GetLatestSnapshot();
+
+	sigprocmask(SIG_SETMASK, &BlockSig, NULL);
+	INSTR_TIME_SET_CURRENT(start);
+
+	while (numregistered < numkeep)
+	{
+		rs[numregistered] = RegisterSnapshot(lsnap);
+		numregistered++;
+	}
+
+	for (i = 0 ; i < numiters; i++)
+	{
+		while (numregistered < numsnaps)
+		{
+			rs[numregistered] = RegisterSnapshot(lsnap);
+			numregistered++;
+		}
+
+		while (numregistered > numkeep)
+		{
+			numregistered--;
+			UnregisterSnapshot(rs[numregistered]);
+		}
+	}
+
+	while (numregistered > 0)
+	{
+		numregistered--;
+		UnregisterSnapshot(rs[numregistered]);
+	}
+
+	INSTR_TIME_SET_CURRENT(duration);
+	INSTR_TIME_SUBTRACT(duration, start);
+	sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);
+
+	PG_RETURN_FLOAT8(INSTR_TIME_GET_MILLISEC(duration));
+};
+
+
+/*
+ * Same, but do the register/unregister operations in
+ * FIFO order.
+ */
+Datum
+snapshotbench_fifo(PG_FUNCTION_ARGS)
+{
+	int			numkeep = PG_GETARG_INT32(0);
+	int			numsnaps = PG_GETARG_INT32(1);
+	int			numiters = PG_GETARG_INT32(2);
+	int			i,
+				j;
+	instr_time	start,
+				duration;
+	Snapshot	lsnap;
+	Snapshot   *rs;
+	int			numregistered = 0;
+
+	rs = palloc(Max(numsnaps, numkeep) * sizeof(Snapshot));
+
+	lsnap = GetLatestSnapshot();
+
+	sigprocmask(SIG_SETMASK, &BlockSig, NULL);
+	INSTR_TIME_SET_CURRENT(start);
+
+	while (numregistered < numkeep)
+	{
+		rs[numregistered] = RegisterSnapshot(lsnap);
+		numregistered++;
+	}
+
+	for (i = 0 ; i < numiters; i++)
+	{
+		while (numregistered < numsnaps)
+		{
+			rs[numregistered] = RegisterSnapshot(lsnap);
+			numregistered++;
+		}
+
+		for (j = numkeep; j < numregistered; j++)
+			UnregisterSnapshot(rs[j]);
+		numregistered = numkeep;
+	}
+
+	for (j = 0; j < numregistered; j++)
+		UnregisterSnapshot(rs[j]);
+	numregistered = numkeep;
+
+	INSTR_TIME_SET_CURRENT(duration);
+	INSTR_TIME_SUBTRACT(duration, start);
+	sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);
+
+	PG_RETURN_FLOAT8(INSTR_TIME_GET_MILLISEC(duration));
+};
diff --git a/contrib/resownerbench/resownerbench.control b/contrib/resownerbench/resownerbench.control
new file mode 100644
index 00000000000..ada88b8eed8
--- /dev/null
+++ b/contrib/resownerbench/resownerbench.control
@@ -0,0 +1,6 @@
+# resownerbench
+
+comment = 'benchmark for ResourceOwners'
+default_version = '1.0'
+module_pathname = '$libdir/resownerbench'
+relocatable = true
diff --git a/contrib/resownerbench/snaptest.sql b/contrib/resownerbench/snaptest.sql
new file mode 100644
index 00000000000..e3c0a9710ee
--- /dev/null
+++ b/contrib/resownerbench/snaptest.sql
@@ -0,0 +1,37 @@
+--
+-- Performance test RegisterSnapshot/UnregisterSnapshot.
+--
+select numkeep, numsnaps,
+       -- numiters,
+       -- round(lifo_time_ms) as lifo_total_time_ms,
+       -- round(fifo_time_ms) as fifo_total_time_ms,
+       round((lifo_time_ms::numeric / (numkeep + (numsnaps - numkeep) * numiters)) * 1000000, 1) as lifo_time_ns,
+       round((fifo_time_ms::numeric / (numkeep + (numsnaps - numkeep) * numiters)) * 1000000, 1) as fifo_time_ns
+from
+(values (0,      1,  10000000 * 10),
+        (0,      5,   2000000 * 10),
+        (0,     10,   1000000 * 10),
+        (0,     60,    100000 * 10),
+        (0,     70,    100000 * 10),
+        (0,    100,    100000 * 10),
+        (0,   1000,     10000 * 10),
+        (0,  10000,      1000 * 10),
+
+-- These tests keep 9 snapshots registered across the iterations. That
+-- exceeds the size of the little array in the patch, so this exercises
+-- the hash lookups. Without the patch, these still fit in the array
+-- (it's 64 entries without the patch)
+        (9,     10,  10000000 * 10),
+        (9,    100,    100000 * 10),
+        (9,   1000,     10000 * 10),
+        (9,  10000,      1000 * 10),
+
+-- These exceed the 64 entry array even without the patch, so these fall
+-- in the hash table regime with and without the patch.
+        (65,    70,   1000000 * 10),
+        (65,   100,    100000 * 10),
+        (65,  1000,     10000 * 10),
+        (65, 10000,      1000 * 10)
+) AS params (numkeep, numsnaps, numiters),
+lateral snapshotbench_lifo(numkeep, numsnaps, numiters) as lifo_time_ms,
+lateral snapshotbench_fifo(numkeep, numsnaps, numiters) as fifo_time_ms;
-- 
2.47.3

From 670fd60c6c953286956a3a2d52cf0b532deaafbb Mon Sep 17 00:00:00 2001
From: Heikki Linnakangas <heikki.linnakangas@iki.fi>
Date: Tue, 21 Oct 2025 09:51:23 +0300
Subject: [PATCH 1/3] resownerbench

---
 contrib/Makefile                             |   1 +
 contrib/meson.build                          |   2 +
 contrib/resownerbench/Makefile               |  17 ++
 contrib/resownerbench/meson.build            |  24 +++
 contrib/resownerbench/resownerbench--1.0.sql |   7 +
 contrib/resownerbench/resownerbench.c        | 154 +++++++++++++++++++
 contrib/resownerbench/resownerbench.control  |   6 +
 contrib/resownerbench/snaptest.sql           |  41 +++++
 8 files changed, 252 insertions(+)
 create mode 100644 contrib/resownerbench/Makefile
 create mode 100644 contrib/resownerbench/meson.build
 create mode 100644 contrib/resownerbench/resownerbench--1.0.sql
 create mode 100644 contrib/resownerbench/resownerbench.c
 create mode 100644 contrib/resownerbench/resownerbench.control
 create mode 100644 contrib/resownerbench/snaptest.sql

diff --git a/contrib/Makefile b/contrib/Makefile
index 2f0a88d3f77..f84b26d52cf 100644
--- a/contrib/Makefile
+++ b/contrib/Makefile
@@ -43,6 +43,7 @@ SUBDIRS = \
 		pg_visibility	\
 		pg_walinspect	\
 		postgres_fdw	\
+		resownerbench	\
 		seg		\
 		spi		\
 		tablefunc	\
diff --git a/contrib/meson.build b/contrib/meson.build
index ed30ee7d639..9c34bb128b7 100644
--- a/contrib/meson.build
+++ b/contrib/meson.build
@@ -71,3 +71,5 @@ subdir('unaccent')
 subdir('uuid-ossp')
 subdir('vacuumlo')
 subdir('xml2')
+
+subdir('resownerbench')
diff --git a/contrib/resownerbench/Makefile b/contrib/resownerbench/Makefile
new file mode 100644
index 00000000000..9b0e1cfee1a
--- /dev/null
+++ b/contrib/resownerbench/Makefile
@@ -0,0 +1,17 @@
+MODULE_big = resownerbench
+OBJS = resownerbench.o
+
+EXTENSION = resownerbench
+DATA = resownerbench--1.0.sql
+PGFILEDESC = "resownerbench - benchmark for ResourceOwners"
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = contrib/resownerbench
+top_builddir = ../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
diff --git a/contrib/resownerbench/meson.build b/contrib/resownerbench/meson.build
new file mode 100644
index 00000000000..30051c3a375
--- /dev/null
+++ b/contrib/resownerbench/meson.build
@@ -0,0 +1,24 @@
+# Copyright (c) 2022-2025, PostgreSQL Global Development Group
+
+resownerbench_sources = files(
+  'resownerbench.c',
+)
+
+resownerbench = shared_module('resownerbench',
+  resownerbench_sources,
+  c_pch: pch_postgres_h,
+  kwargs: contrib_mod_args,
+)
+contrib_targets += resownerbench
+
+install_data(
+  'resownerbench.control',
+  'resownerbench--1.0.sql',
+  kwargs: contrib_data_args,
+)
+
+tests += {
+  'name': 'resownerbench',
+  'sd': meson.current_source_dir(),
+  'bd': meson.current_build_dir(),
+}
diff --git a/contrib/resownerbench/resownerbench--1.0.sql b/contrib/resownerbench/resownerbench--1.0.sql
new file mode 100644
index 00000000000..8521caed2d8
--- /dev/null
+++ b/contrib/resownerbench/resownerbench--1.0.sql
@@ -0,0 +1,7 @@
+CREATE OR REPLACE FUNCTION snapshotbench_fifo(numkeep int, numsnaps int, numiters int) RETURNS double precision
+AS 'resownerbench', 'snapshotbench_fifo'
+LANGUAGE C;
+
+CREATE OR REPLACE FUNCTION snapshotbench_lifo(numkeep int, numsnaps int, numiters int) RETURNS double precision
+AS 'resownerbench', 'snapshotbench_lifo'
+LANGUAGE C;
diff --git a/contrib/resownerbench/resownerbench.c b/contrib/resownerbench/resownerbench.c
new file mode 100644
index 00000000000..80f98c2105b
--- /dev/null
+++ b/contrib/resownerbench/resownerbench.c
@@ -0,0 +1,154 @@
+#include "postgres.h"
+
+#include "catalog/pg_type.h"
+#include "catalog/pg_statistic.h"
+#include "executor/spi.h"
+#include "funcapi.h"
+#include "libpq/pqsignal.h"
+#include "utils/catcache.h"
+#include "utils/syscache.h"
+#include "utils/timestamp.h"
+#include "utils/snapmgr.h"
+
+PG_MODULE_MAGIC;
+
+PG_FUNCTION_INFO_V1(snapshotbench_lifo);
+PG_FUNCTION_INFO_V1(snapshotbench_fifo);
+
+/*
+ * ResourceOwner Performance test, using RegisterSnapshot().
+ *
+ * This takes three parameters: numkeep, numsnaps, numiters.
+ *
+ * First, we register 'numkeep' snapshots. They are kept registed
+ * until the end of the test. Then, we repeatedly register and
+ * unregister 'numsnaps - numkeep' additional snapshots, repeating
+ * 'numiters' times. All the register/unregister calls are made in
+ * LIFO order.
+ *
+ * Returns the time spent, in milliseconds.
+ *
+ * The idea is to test the performance of ResourceOwnerRemember()
+ * and ReourceOwnerForget() operations, under different regimes.
+ *
+ * In the old implementation, if 'numsnaps' is small enough, all
+ * the entries fit in the resource owner's small array (it can
+ * hold 64 entries).
+ *
+ * In the new implementation, the array is much smaller, only 8
+ * entries, but it's used together with the hash table so that
+ * we stay in the "array regime" as long as 'numsnaps - numkeep'
+ * is smaller than 8 entries.
+ *
+ * 'numiters' can be adjusted to adjust the overall runtime to be
+ * suitable long.
+ */
+Datum
+snapshotbench_lifo(PG_FUNCTION_ARGS)
+{
+	int			numkeep = PG_GETARG_INT32(0);
+	int			numsnaps = PG_GETARG_INT32(1);
+	int			numiters = PG_GETARG_INT32(2);
+	int			i;
+	instr_time	start,
+				duration;
+	Snapshot	lsnap;
+	Snapshot   *rs;
+	int			numregistered = 0;
+
+	rs = palloc(Max(numsnaps, numkeep) * sizeof(Snapshot));
+
+	lsnap = GetLatestSnapshot();
+
+	sigprocmask(SIG_SETMASK, &BlockSig, NULL);
+	INSTR_TIME_SET_CURRENT(start);
+
+	while (numregistered < numkeep)
+	{
+		rs[numregistered] = RegisterSnapshot(lsnap);
+		numregistered++;
+	}
+
+	for (i = 0 ; i < numiters; i++)
+	{
+		while (numregistered < numsnaps)
+		{
+			rs[numregistered] = RegisterSnapshot(lsnap);
+			numregistered++;
+		}
+
+		while (numregistered > numkeep)
+		{
+			numregistered--;
+			UnregisterSnapshot(rs[numregistered]);
+		}
+	}
+
+	while (numregistered > 0)
+	{
+		numregistered--;
+		UnregisterSnapshot(rs[numregistered]);
+	}
+
+	INSTR_TIME_SET_CURRENT(duration);
+	INSTR_TIME_SUBTRACT(duration, start);
+	sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);
+
+	PG_RETURN_FLOAT8(INSTR_TIME_GET_MILLISEC(duration));
+};
+
+
+/*
+ * Same, but do the register/unregister operations in
+ * FIFO order.
+ */
+Datum
+snapshotbench_fifo(PG_FUNCTION_ARGS)
+{
+	int			numkeep = PG_GETARG_INT32(0);
+	int			numsnaps = PG_GETARG_INT32(1);
+	int			numiters = PG_GETARG_INT32(2);
+	int			i,
+				j;
+	instr_time	start,
+				duration;
+	Snapshot	lsnap;
+	Snapshot   *rs;
+	int			numregistered = 0;
+
+	rs = palloc(Max(numsnaps, numkeep) * sizeof(Snapshot));
+
+	lsnap = GetLatestSnapshot();
+
+	sigprocmask(SIG_SETMASK, &BlockSig, NULL);
+	INSTR_TIME_SET_CURRENT(start);
+
+	while (numregistered < numkeep)
+	{
+		rs[numregistered] = RegisterSnapshot(lsnap);
+		numregistered++;
+	}
+
+	for (i = 0 ; i < numiters; i++)
+	{
+		while (numregistered < numsnaps)
+		{
+			rs[numregistered] = RegisterSnapshot(lsnap);
+			numregistered++;
+		}
+
+		for (j = numkeep; j < numregistered; j++)
+			UnregisterSnapshot(rs[j]);
+		numregistered = numkeep;
+	}
+
+	for (j = 0; j < numregistered; j++)
+		UnregisterSnapshot(rs[j]);
+	numregistered = numkeep;
+
+	INSTR_TIME_SET_CURRENT(duration);
+	INSTR_TIME_SUBTRACT(duration, start);
+	sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);
+
+	PG_RETURN_FLOAT8(INSTR_TIME_GET_MILLISEC(duration));
+};
diff --git a/contrib/resownerbench/resownerbench.control b/contrib/resownerbench/resownerbench.control
new file mode 100644
index 00000000000..ada88b8eed8
--- /dev/null
+++ b/contrib/resownerbench/resownerbench.control
@@ -0,0 +1,6 @@
+# resownerbench
+
+comment = 'benchmark for ResourceOwners'
+default_version = '1.0'
+module_pathname = '$libdir/resownerbench'
+relocatable = true
diff --git a/contrib/resownerbench/snaptest.sql b/contrib/resownerbench/snaptest.sql
new file mode 100644
index 00000000000..a8fd4da2f52
--- /dev/null
+++ b/contrib/resownerbench/snaptest.sql
@@ -0,0 +1,41 @@
+--
+-- Performance test RegisterSnapshot/UnregisterSnapshot.
+--
+with params as materialized (
+select * from
+generate_series(1,10) s(run),
+(values (0,      1,  10000000 * 10),
+        (0,      5,   2000000 * 10),
+        (0,     10,   1000000 * 10),
+        (0,     60,    100000 * 10),
+        (0,     70,    100000 * 10),
+        (0,    100,    100000 * 10),
+        (0,   1000,     10000 * 10),
+        (0,  10000,      1000 * 10),
+
+-- These tests keep 9 snapshots registered across the iterations. That
+-- exceeds the size of the little array in the patch, so this exercises
+-- the hash lookups. Without the patch, these still fit in the array
+-- (it's 64 entries without the patch)
+        (9,     10,  10000000 * 10),
+        (9,    100,    100000 * 10),
+        (9,   1000,     10000 * 10),
+        (9,  10000,      1000 * 10),
+
+-- These exceed the 64 entry array even without the patch, so these fall
+-- in the hash table regime with and without the patch.
+        (65,    70,   1000000 * 10),
+        (65,   100,    100000 * 10),
+        (65,  1000,     10000 * 10),
+        (65, 10000,      1000 * 10)
+) AS params (numkeep, numsnaps, numiters)
+)
+select run, numkeep, numsnaps,
+       -- numiters,
+       -- round(lifo_time_ms) as lifo_total_time_ms,
+       -- round(fifo_time_ms) as fifo_total_time_ms,
+       round((lifo_time_ms::numeric / (numkeep + (numsnaps - numkeep) * numiters)) * 1000000, 1) as lifo_time_ns,
+       round((fifo_time_ms::numeric / (numkeep + (numsnaps - numkeep) * numiters)) * 1000000, 1) as fifo_time_ns
+from params,
+lateral snapshotbench_lifo(numkeep, numsnaps, numiters) as lifo_time_ms,
+lateral snapshotbench_fifo(numkeep, numsnaps, numiters) as fifo_time_ms;
-- 
2.51.0

From 83ff6c494312ea778172c446c3c4ca042dbb3d67 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <tomas@vondra.me>
Date: Thu, 16 Oct 2025 13:27:25 +0200
Subject: [PATCH 2/3] v2: Deduplicate entries in ResourceOwner

We may track the same resource many times, but the current ResourceOwner
does not handle that efficiently. Each reference is a separate entry,
which leads to long linear searches in the simple hash table.

This adds a simple opportunistic deduplication. Entries get a counter,
tracking how many references they represent. It only applies to the
"hash" phase, not to the initial array. And when adding an entry to the
hash finds an empty slot first, it uses it (instead of continuing to
look for a possible match).

This means the deduplication is not perfect - there may be multiple
entries for the same value. But in practice it seems like a good trade
off. It won't affect cases with no duplicates, and it will help cases
with many of them.

The code also continues to grow the hash table as before. In some cases
it might be enough to deduplicate the contents. But it's hard to say in
advance, and it's not much cheaper as it still requires walking the
current hash table. So there's a risk of attempting deduplication, only
to find the hash still needs to be enlarged, doubling the cost. So we
just grow the hash table. In the worst case the memory usage is not
worse than without deduplication. If deduplication is effective, the
hash table grows later or not at all.
---
 src/backend/utils/resowner/resowner.c | 215 ++++++++++++++++++++------
 1 file changed, 166 insertions(+), 49 deletions(-)

diff --git a/src/backend/utils/resowner/resowner.c b/src/backend/utils/resowner/resowner.c
index fca84ded6dd..73de45baccd 100644
--- a/src/backend/utils/resowner/resowner.c
+++ b/src/backend/utils/resowner/resowner.c
@@ -15,6 +15,12 @@
  * a particular reference, you need to search both the array and the hash
  * table.
  *
+ * The hash table deduplicates entries in an opportunistic way. If it finds
+ * an entry for the same resource (same kind), it increments a counter instead.
+ * But if it finds an empty slot first, it uses the slot. If the table gets
+ * enlarged, those entries will be merged, but it's possible to have multiple
+ * entries for the same resource. The fixed-size array does not deduplicate.
+ *
  * The most frequent usage is that a resource is remembered, and forgotten
  * shortly thereafter.  For example, pin a buffer, read one tuple from it,
  * release the pin.  Linearly scanning the small array handles that case
@@ -67,6 +73,18 @@ typedef struct ResourceElem
 	const ResourceOwnerDesc *kind;	/* NULL indicates a free hash table slot */
 } ResourceElem;
 
+/*
+ * ResourceHashElem represents a reference associated with a resource owner,
+ * stored in the hash table. Similar to ResourceElem, but has a counter field
+ * tracking how many references it represents, to allow deduplication.
+ */
+typedef struct ResourceHashElem
+{
+	Datum		item;
+	uint32		count;				/* number of references */
+	const ResourceOwnerDesc *kind;	/* NULL indicates a free hash table slot */
+} ResourceHashElem;
+
 /*
  * Size of the fixed-size array to hold most-recently remembered resources.
  */
@@ -151,7 +169,7 @@ struct ResourceOwnerData
 	 * release priority.  The first 'nhash' elements are occupied, the rest
 	 * are empty.
 	 */
-	ResourceElem *hash;
+	ResourceHashElem *hash;
 	uint32		capacity;		/* allocated length of hash[] */
 	uint32		grow_at;		/* grow hash when reach this */
 
@@ -198,7 +216,8 @@ static ResourceReleaseCallbackItem *ResourceRelease_callbacks = NULL;
 /* Internal routines */
 static inline uint32 hash_resource_elem(Datum value, const ResourceOwnerDesc *kind);
 static void ResourceOwnerAddToHash(ResourceOwner owner, Datum value,
-								   const ResourceOwnerDesc *kind);
+								   const ResourceOwnerDesc *kind,
+								   uint32 count);
 static int	resource_priority_cmp(const void *a, const void *b);
 static void ResourceOwnerSort(ResourceOwner owner);
 static void ResourceOwnerReleaseAll(ResourceOwner owner,
@@ -239,7 +258,8 @@ hash_resource_elem(Datum value, const ResourceOwnerDesc *kind)
  * Adds 'value' of given 'kind' to the ResourceOwner's hash table
  */
 static void
-ResourceOwnerAddToHash(ResourceOwner owner, Datum value, const ResourceOwnerDesc *kind)
+ResourceOwnerAddToHash(ResourceOwner owner, Datum value, const ResourceOwnerDesc *kind,
+					   uint32 count)
 {
 	uint32		mask = owner->capacity - 1;
 	uint32		idx;
@@ -250,12 +270,25 @@ ResourceOwnerAddToHash(ResourceOwner owner, Datum value, const ResourceOwnerDesc
 	idx = hash_resource_elem(value, kind) & mask;
 	for (;;)
 	{
+		/*
+		 * Increment counter for a matching entry or use an empty slot,
+		 * whichever we find first.
+		 */
+
+		if ((owner->hash[idx].kind == kind) &&
+			(owner->hash[idx].item == value))
+		{
+			owner->hash[idx].count += count;
+			return;
+		}
+
 		if (owner->hash[idx].kind == NULL)
 			break;				/* found a free slot */
 		idx = (idx + 1) & mask;
 	}
 	owner->hash[idx].item = value;
 	owner->hash[idx].kind = kind;
+	owner->hash[idx].count = count;
 	owner->nhash++;
 }
 
@@ -277,6 +310,24 @@ resource_priority_cmp(const void *a, const void *b)
 		return 1;
 }
 
+/*
+ * Comparison function to sort by release phase and priority
+ */
+static int
+resource_priority_hash_cmp(const void *a, const void *b)
+{
+	const ResourceHashElem *ra = (const ResourceHashElem *) a;
+	const ResourceHashElem *rb = (const ResourceHashElem *) b;
+
+	/* Note: reverse order */
+	if (ra->kind->release_phase == rb->kind->release_phase)
+		return pg_cmp_u32(rb->kind->release_priority, ra->kind->release_priority);
+	else if (ra->kind->release_phase > rb->kind->release_phase)
+		return -1;
+	else
+		return 1;
+}
+
 /*
  * Sort resources in reverse release priority.
  *
@@ -288,16 +339,21 @@ resource_priority_cmp(const void *a, const void *b)
 static void
 ResourceOwnerSort(ResourceOwner owner)
 {
-	ResourceElem *items;
-	uint32		nitems;
-
 	if (owner->nhash == 0)
 	{
+		ResourceElem *items;
+		uint32		nitems;
+
 		items = owner->arr;
 		nitems = owner->narr;
+
+		qsort(items, nitems, sizeof(ResourceElem), resource_priority_cmp);
 	}
 	else
 	{
+		ResourceHashElem   *items;
+		uint32				nitems;
+
 		/*
 		 * Compact the hash table, so that all the elements are in the
 		 * beginning of the 'hash' array, with no empty elements.
@@ -324,7 +380,9 @@ ResourceOwnerSort(ResourceOwner owner)
 		Assert(dst + owner->narr <= owner->capacity);
 		for (int idx = 0; idx < owner->narr; idx++)
 		{
-			owner->hash[dst] = owner->arr[idx];
+			owner->hash[dst].item = owner->arr[idx].item;
+			owner->hash[dst].kind = owner->arr[idx].kind;
+			owner->hash[dst].count = 1;
 			dst++;
 		}
 		Assert(dst == owner->nhash + owner->narr);
@@ -333,9 +391,9 @@ ResourceOwnerSort(ResourceOwner owner)
 
 		items = owner->hash;
 		nitems = owner->nhash;
-	}
 
-	qsort(items, nitems, sizeof(ResourceElem), resource_priority_cmp);
+		qsort(items, nitems, sizeof(ResourceHashElem), resource_priority_hash_cmp);
+	}
 }
 
 /*
@@ -345,9 +403,6 @@ static void
 ResourceOwnerReleaseAll(ResourceOwner owner, ResourceReleasePhase phase,
 						bool printLeakWarnings)
 {
-	ResourceElem *items;
-	uint32		nitems;
-
 	/*
 	 * ResourceOwnerSort must've been called already.  All the resources are
 	 * either in the array or the hash.
@@ -356,49 +411,93 @@ ResourceOwnerReleaseAll(ResourceOwner owner, ResourceReleasePhase phase,
 	Assert(owner->sorted);
 	if (owner->nhash == 0)
 	{
+		ResourceElem *items;
+		uint32		nitems;
+
 		items = owner->arr;
 		nitems = owner->narr;
+
+		/*
+		 * The resources are sorted in reverse priority order.  Release them
+		 * starting from the end, until we hit the end of the phase that we are
+		 * releasing now.  We will continue from there when called again for the
+		 * next phase.
+		 */
+		while (nitems > 0)
+		{
+			uint32		idx = nitems - 1;
+			Datum		value = items[idx].item;
+			const ResourceOwnerDesc *kind = items[idx].kind;
+
+			if (kind->release_phase > phase)
+				break;
+			Assert(kind->release_phase == phase);
+
+			if (printLeakWarnings)
+			{
+				char	   *res_str;
+
+				res_str = kind->DebugPrint ?
+					kind->DebugPrint(value)
+					: psprintf("%s %p", kind->name, DatumGetPointer(value));
+				pfree(res_str);
+			}
+
+			kind->ReleaseResource(value);
+
+			nitems--;
+		}
+
+		owner->narr = nitems;
 	}
 	else
 	{
+		ResourceHashElem *items;
+		uint32		nitems;
+
 		Assert(owner->narr == 0);
 		items = owner->hash;
 		nitems = owner->nhash;
-	}
 
-	/*
-	 * The resources are sorted in reverse priority order.  Release them
-	 * starting from the end, until we hit the end of the phase that we are
-	 * releasing now.  We will continue from there when called again for the
-	 * next phase.
-	 */
-	while (nitems > 0)
-	{
-		uint32		idx = nitems - 1;
-		Datum		value = items[idx].item;
-		const ResourceOwnerDesc *kind = items[idx].kind;
+		/*
+		 * The resources are sorted in reverse priority order.  Release them
+		 * starting from the end, until we hit the end of the phase that we are
+		 * releasing now.  We will continue from there when called again for the
+		 * next phase.
+		 */
+		while (nitems > 0)
+		{
+			uint32		idx = nitems - 1;
+			Datum		value = items[idx].item;
+			const ResourceOwnerDesc *kind = items[idx].kind;
+			uint32		count = items[idx].count;
 
-		if (kind->release_phase > phase)
-			break;
-		Assert(kind->release_phase == phase);
+			if (kind->release_phase > phase)
+				break;
+			Assert(kind->release_phase == phase);
 
-		if (printLeakWarnings)
-		{
-			char	   *res_str;
+			if (printLeakWarnings)
+			{
+				char	   *res_str;
+
+				res_str = kind->DebugPrint ?
+					kind->DebugPrint(value)
+					: psprintf("%s %p", kind->name, DatumGetPointer(value));
+				pfree(res_str);
+			}
 
-			res_str = kind->DebugPrint ?
-				kind->DebugPrint(value)
-				: psprintf("%s %p", kind->name, DatumGetPointer(value));
-			elog(WARNING, "resource was not closed: %s", res_str);
-			pfree(res_str);
+			/* invoke the release callback for each reference */
+			while (count > 0)
+			{
+				kind->ReleaseResource(value);
+				count--;
+			}
+
+			nitems--;
 		}
-		kind->ReleaseResource(value);
-		nitems--;
-	}
-	if (owner->nhash == 0)
-		owner->narr = nitems;
-	else
+
 		owner->nhash = nitems;
+	}
 }
 
 
@@ -466,16 +565,16 @@ ResourceOwnerEnlarge(ResourceOwner owner)
 		uint32		i,
 					oldcap,
 					newcap;
-		ResourceElem *oldhash;
-		ResourceElem *newhash;
+		ResourceHashElem *oldhash;
+		ResourceHashElem *newhash;
 
 		oldhash = owner->hash;
 		oldcap = owner->capacity;
 
 		/* Double the capacity (it must stay a power of 2!) */
 		newcap = (oldcap > 0) ? oldcap * 2 : RESOWNER_HASH_INIT_SIZE;
-		newhash = (ResourceElem *) MemoryContextAllocZero(TopMemoryContext,
-														  newcap * sizeof(ResourceElem));
+		newhash = (ResourceHashElem *) MemoryContextAllocZero(TopMemoryContext,
+														  newcap * sizeof(ResourceHashElem));
 
 		/*
 		 * We assume we can't fail below this point, so OK to scribble on the
@@ -496,7 +595,8 @@ ResourceOwnerEnlarge(ResourceOwner owner)
 			for (i = 0; i < oldcap; i++)
 			{
 				if (oldhash[i].kind != NULL)
-					ResourceOwnerAddToHash(owner, oldhash[i].item, oldhash[i].kind);
+					ResourceOwnerAddToHash(owner, oldhash[i].item, oldhash[i].kind,
+										   oldhash[i].count);
 			}
 
 			/* And release old hash table. */
@@ -506,7 +606,7 @@ ResourceOwnerEnlarge(ResourceOwner owner)
 
 	/* Move items from the array to the hash */
 	for (int i = 0; i < owner->narr; i++)
-		ResourceOwnerAddToHash(owner, owner->arr[i].item, owner->arr[i].kind);
+		ResourceOwnerAddToHash(owner, owner->arr[i].item, owner->arr[i].kind, 1);
 	owner->narr = 0;
 
 	Assert(owner->nhash <= owner->grow_at);
@@ -555,7 +655,8 @@ ResourceOwnerRemember(ResourceOwner owner, Datum value, const ResourceOwnerDesc
  * Note: Forgetting a resource does not guarantee that there is room to
  * remember a new resource.  One exception is when you forget the most
  * recently remembered resource; that does make room for a new remember call.
- * Some code callers rely on that exception.
+ * Some code callers rely on that exception. The deduplication does not
+ * change this, as it only applies to the hash table.
  */
 void
 ResourceOwnerForget(ResourceOwner owner, Datum value, const ResourceOwnerDesc *kind)
@@ -597,8 +698,17 @@ ResourceOwnerForget(ResourceOwner owner, Datum value, const ResourceOwnerDesc *k
 			if (owner->hash[idx].item == value &&
 				owner->hash[idx].kind == kind)
 			{
+				/* decrement the count for entry for multiple references */
+				if (owner->hash[idx].count > 1)
+				{
+					owner->hash[idx].count--;
+					return;
+				}
+
+				/* remove the entry entirely */
 				owner->hash[idx].item = (Datum) 0;
 				owner->hash[idx].kind = NULL;
+				owner->hash[idx].count = 0;
 				owner->nhash--;
 
 #ifdef RESOWNER_STATS
@@ -847,12 +957,19 @@ ResourceOwnerReleaseAllOfKind(ResourceOwner owner, const ResourceOwnerDesc *kind
 		if (owner->hash[i].kind == kind)
 		{
 			Datum		value = owner->hash[i].item;
+			uint32		count = owner->hash[i].count;
 
 			owner->hash[i].item = (Datum) 0;
 			owner->hash[i].kind = NULL;
+			owner->hash[i].count = 0;
 			owner->nhash--;
 
-			kind->ReleaseResource(value);
+			/* invoke the release callback once for each reference */
+			while (count > 0)
+			{
+				kind->ReleaseResource(value);
+				count--;
+			}
 		}
 	}
 	owner->releasing = false;
-- 
2.51.0

From 98efe8f0a3ba4474bf21af6e7e727d529bb89d45 Mon Sep 17 00:00:00 2001
From: tomas <tomas>
Date: Tue, 21 Oct 2025 15:06:01 +0200
Subject: [PATCH 3/3] v3: Check for empty slots first

---
 src/backend/utils/resowner/resowner.c | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/src/backend/utils/resowner/resowner.c b/src/backend/utils/resowner/resowner.c
index 73de45baccd..1a04eec945a 100644
--- a/src/backend/utils/resowner/resowner.c
+++ b/src/backend/utils/resowner/resowner.c
@@ -270,11 +270,13 @@ ResourceOwnerAddToHash(ResourceOwner owner, Datum value, const ResourceOwnerDesc
 	idx = hash_resource_elem(value, kind) & mask;
 	for (;;)
 	{
+		if (owner->hash[idx].kind == NULL)
+			break;				/* found a free slot */
+
 		/*
 		 * Increment counter for a matching entry or use an empty slot,
 		 * whichever we find first.
 		 */
-
 		if ((owner->hash[idx].kind == kind) &&
 			(owner->hash[idx].item == value))
 		{
@@ -282,8 +284,6 @@ ResourceOwnerAddToHash(ResourceOwner owner, Datum value, const ResourceOwnerDesc
 			return;
 		}
 
-		if (owner->hash[idx].kind == NULL)
-			break;				/* found a free slot */
 		idx = (idx + 1) & mask;
 	}
 	owner->hash[idx].item = value;
-- 
2.51.0

From b9110bbad383ce32d4dfe1ad03c44138c0d53c09 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <tomas@vondra.me>
Date: Mon, 27 Oct 2025 19:02:10 +0100
Subject: [PATCH] introduce batch of slots

---
 src/backend/access/table/tableam.c | 36 +++++++++++++
 src/backend/commands/copyfrom.c    | 15 ++++--
 src/backend/executor/execTuples.c  | 84 ++++++++++++++++++++++++++++++
 src/include/access/tableam.h       |  3 ++
 src/include/executor/tuptable.h    | 16 ++++++
 5 files changed, 150 insertions(+), 4 deletions(-)

diff --git a/src/backend/access/table/tableam.c b/src/backend/access/table/tableam.c
index 5e41404937e..a3e67940443 100644
--- a/src/backend/access/table/tableam.c
+++ b/src/backend/access/table/tableam.c
@@ -103,6 +103,42 @@ table_slot_create(Relation relation, List **reglist)
 	return slot;
 }
 
+TupleTableSlotBatch *
+table_slot_batch_create(Relation relation, int maxslots)
+{
+	TupleTableSlotBatch *batch;
+
+	batch = palloc0(offsetof(TupleTableSlotBatch, ttsb_slots) +
+					maxslots * sizeof(TupleTableSlot *));
+
+	batch->ttsb_maxslots = maxslots;
+	batch->ttsb_nslots = 0;
+
+	/* const for optimization purposes, OK to modify at allocation time */
+	*((const TupleTableSlotOps **) &batch->ttsb_ops) = table_slot_callbacks(relation);
+	batch->ttsb_tupleDescriptor = RelationGetDescr(relation);
+
+	PinTupleDesc(batch->ttsb_tupleDescriptor);
+
+	return batch;
+}
+
+TupleTableSlot *
+table_slot_batch_next(TupleTableSlotBatch *batch)
+{
+	TupleTableSlot *slot;
+
+	if (batch->ttsb_nslots == batch->ttsb_maxslots)
+		elog(ERROR, "not enough slots in the batch");
+
+	slot = MakeSingleTupleTableSlotBatch(batch->ttsb_tupleDescriptor,
+										 batch->ttsb_ops);
+
+	batch->ttsb_slots[batch->ttsb_nslots++] = slot;
+
+	return slot;
+}
+
 
 /* ----------------------------------------------------------------------------
  * Table scan functions.
diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c
index 12781963b4f..e435b56fbe1 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -77,6 +77,7 @@
 /* Stores multi-insert data related to a single relation in CopyFrom. */
 typedef struct CopyMultiInsertBuffer
 {
+	TupleTableSlotBatch *batch;
 	TupleTableSlot *slots[MAX_BUFFERED_TUPLES]; /* Array to store tuples */
 	ResultRelInfo *resultRelInfo;	/* ResultRelInfo for 'relid' */
 	BulkInsertState bistate;	/* BulkInsertState for this rel if plain
@@ -369,6 +370,7 @@ CopyMultiInsertBufferInit(ResultRelInfo *rri)
 	buffer->resultRelInfo = rri;
 	buffer->bistate = (rri->ri_FdwRoutine == NULL) ? GetBulkInsertState() : NULL;
 	buffer->nused = 0;
+	buffer->batch = NULL;
 
 	return buffer;
 }
@@ -621,7 +623,7 @@ CopyMultiInsertBufferCleanup(CopyMultiInsertInfo *miinfo,
 							 CopyMultiInsertBuffer *buffer)
 {
 	ResultRelInfo *resultRelInfo = buffer->resultRelInfo;
-	int			i;
+	// int			i;
 
 	/* Ensure buffer was flushed */
 	Assert(buffer->nused == 0);
@@ -638,8 +640,10 @@ CopyMultiInsertBufferCleanup(CopyMultiInsertInfo *miinfo,
 		Assert(buffer->bistate == NULL);
 
 	/* Since we only create slots on demand, just drop the non-null ones. */
-	for (i = 0; i < MAX_BUFFERED_TUPLES && buffer->slots[i] != NULL; i++)
-		ExecDropSingleTupleTableSlot(buffer->slots[i]);
+	// for (i = 0; i < MAX_BUFFERED_TUPLES && buffer->slots[i] != NULL; i++)
+	//	ExecDropSingleTupleTableSlot(buffer->slots[i]);
+	if (buffer->batch != NULL)
+		ExecDropTupleTableSlotBatch(buffer->batch);
 
 	if (resultRelInfo->ri_FdwRoutine == NULL)
 		table_finish_bulk_insert(resultRelInfo->ri_RelationDesc,
@@ -743,8 +747,11 @@ CopyMultiInsertInfoNextFreeSlot(CopyMultiInsertInfo *miinfo,
 
 	nused = buffer->nused;
 
+	if (buffer->batch == NULL)
+		buffer->batch = table_slot_batch_create(rri->ri_RelationDesc, MAX_BUFFERED_TUPLES);
+
 	if (buffer->slots[nused] == NULL)
-		buffer->slots[nused] = table_slot_create(rri->ri_RelationDesc, NULL);
+		buffer->slots[nused] = table_slot_batch_next(buffer->batch);
 	return buffer->slots[nused];
 }
 
diff --git a/src/backend/executor/execTuples.c b/src/backend/executor/execTuples.c
index 8e02d68824f..b1f9435ae51 100644
--- a/src/backend/executor/execTuples.c
+++ b/src/backend/executor/execTuples.c
@@ -1350,6 +1350,57 @@ MakeTupleTableSlot(TupleDesc tupleDesc,
 	return slot;
 }
 
+TupleTableSlot *
+MakeTupleTableSlotBatch(TupleDesc tupleDesc,
+						const TupleTableSlotOps *tts_ops)
+{
+	Size		basesz,
+				allocsz;
+	TupleTableSlot *slot;
+
+	basesz = tts_ops->base_slot_size;
+
+	/*
+	 * When a fixed descriptor is specified, we can reduce overhead by
+	 * allocating the entire slot in one go.
+	 */
+	if (tupleDesc)
+		allocsz = MAXALIGN(basesz) +
+			MAXALIGN(tupleDesc->natts * sizeof(Datum)) +
+			MAXALIGN(tupleDesc->natts * sizeof(bool));
+	else
+		allocsz = basesz;
+
+	slot = palloc0(allocsz);
+	/* const for optimization purposes, OK to modify at allocation time */
+	*((const TupleTableSlotOps **) &slot->tts_ops) = tts_ops;
+	slot->type = T_TupleTableSlot;
+	slot->tts_flags |= TTS_FLAG_EMPTY;
+	if (tupleDesc != NULL)
+		slot->tts_flags |= TTS_FLAG_FIXED;
+	slot->tts_tupleDescriptor = tupleDesc;
+	slot->tts_mcxt = CurrentMemoryContext;
+	slot->tts_nvalid = 0;
+
+	if (tupleDesc != NULL)
+	{
+		slot->tts_values = (Datum *)
+			(((char *) slot)
+			 + MAXALIGN(basesz));
+		slot->tts_isnull = (bool *)
+			(((char *) slot)
+			 + MAXALIGN(basesz)
+			 + MAXALIGN(tupleDesc->natts * sizeof(Datum)));
+	}
+
+	/*
+	 * And allow slot type specific initialization.
+	 */
+	slot->tts_ops->init(slot);
+
+	return slot;
+}
+
 /* --------------------------------
  *		ExecAllocTableSlot
  *
@@ -1458,6 +1509,39 @@ ExecDropSingleTupleTableSlot(TupleTableSlot *slot)
 	pfree(slot);
 }
 
+TupleTableSlot *
+MakeSingleTupleTableSlotBatch(TupleDesc tupdesc,
+							  const TupleTableSlotOps *tts_ops)
+{
+	TupleTableSlot *slot = MakeTupleTableSlotBatch(tupdesc, tts_ops);
+
+	return slot;
+}
+
+/* FIXME Does this need to match ExecResetTupleTable's processing of one slot? */
+void
+ExecDropTupleTableSlotBatch(TupleTableSlotBatch *batch)
+{
+	for (int i = 0; i < batch->ttsb_nslots; i++)
+	{
+		TupleTableSlot *slot = batch->ttsb_slots[i];
+
+		Assert(IsA(slot, TupleTableSlot));
+		ExecClearTuple(slot);
+		slot->tts_ops->release(slot);
+		if (!TTS_FIXED(slot))
+		{
+			if (slot->tts_values)
+				pfree(slot->tts_values);
+			if (slot->tts_isnull)
+				pfree(slot->tts_isnull);
+		}
+		pfree(slot);
+	}
+
+	ReleaseTupleDesc(batch->ttsb_tupleDescriptor);
+}
+
 
 /* ----------------------------------------------------------------
  *				  tuple table slot accessor functions
diff --git a/src/include/access/tableam.h b/src/include/access/tableam.h
index e16bf025692..b9500db1129 100644
--- a/src/include/access/tableam.h
+++ b/src/include/access/tableam.h
@@ -862,6 +862,9 @@ extern const TupleTableSlotOps *table_slot_callbacks(Relation relation);
  */
 extern TupleTableSlot *table_slot_create(Relation relation, List **reglist);
 
+extern TupleTableSlotBatch *table_slot_batch_create(Relation relation, int maxslots);
+extern TupleTableSlot *table_slot_batch_next(TupleTableSlotBatch *batch);
+
 
 /* ----------------------------------------------------------------------------
  * Table scan functions.
diff --git a/src/include/executor/tuptable.h b/src/include/executor/tuptable.h
index 43f1d999b91..38a8e170c89 100644
--- a/src/include/executor/tuptable.h
+++ b/src/include/executor/tuptable.h
@@ -225,6 +225,17 @@ struct TupleTableSlotOps
 	MinimalTuple (*copy_minimal_tuple) (TupleTableSlot *slot, Size extra);
 };
 
+/* a batch of tuple table slots */
+typedef struct TupleTableSlotBatch
+{
+	NodeTag		type;
+	int			ttsb_maxslots;
+	int			ttsb_nslots;
+	const TupleTableSlotOps *const ttsb_ops; /* implementation of slot */
+	TupleDesc		ttsb_tupleDescriptor;	/* slot's tuple descriptor */
+	TupleTableSlot *ttsb_slots[FLEXIBLE_ARRAY_MEMBER];
+} TupleTableSlotBatch;
+
 /*
  * Predefined TupleTableSlotOps for various types of TupleTableSlotOps. The
  * same are used to identify the type of a given slot.
@@ -347,6 +358,11 @@ extern void slot_getmissingattrs(TupleTableSlot *slot, int startAttNum,
 								 int lastAttNum);
 extern void slot_getsomeattrs_int(TupleTableSlot *slot, int attnum);
 
+extern TupleTableSlot *MakeTupleTableSlotBatch(TupleDesc tupleDesc,
+											   const TupleTableSlotOps *tts_ops);
+extern TupleTableSlot *MakeSingleTupleTableSlotBatch(TupleDesc tupleDesc,
+													 const TupleTableSlotOps *tts_ops);
+extern void ExecDropTupleTableSlotBatch(TupleTableSlotBatch *batch);
 
 #ifndef FRONTEND
 
-- 
2.51.0