diff --git a/src/backend/utils/activity/pgstat.c b/src/backend/utils/activity/pgstat.c
index 0b44a53ebe9..c393bf7d59e 100644
--- a/src/backend/utils/activity/pgstat.c
+++ b/src/backend/utils/activity/pgstat.c
@@ -18,10 +18,10 @@
  * Fixed-numbered stats are stored in plain (non-dynamic) shared memory.
  *
  * Statistics for variable-numbered objects are stored in dynamic shared
- * memory and can be found via a dshash hashtable. The statistics counters are
- * not part of the dshash entry (PgStatShared_HashEntry) directly, but are
- * separately allocated (PgStatShared_HashEntry->body). The separate
- * allocation allows different kinds of statistics to be stored in the same
+ * memory and can be found via shared stats dshash hashtables. The statistics
+ * counters are not part of the dshash entry (PgStatShared_HashEntry)
+ * directly, but are separately allocated (PgStatShared_HashEntry->body). The
+ * separate allocation allows different kinds of statistics to share a
  * hashtable without wasting space in PgStatShared_HashEntry.
  *
  * Variable-numbered stats are addressed by PgStat_HashKey while running.  It
@@ -29,14 +29,13 @@
  * that way at runtime. A wider identifier can be used when serializing to
  * disk (used for replication slot stats).
  *
- * To avoid contention on the shared hashtable, each backend has a
- * backend-local hashtable (pgStatEntryRefHash) in front of the shared
- * hashtable, containing references (PgStat_EntryRef) to shared hashtable
- * entries. The shared hashtable only needs to be accessed when no prior
- * reference is found in the local hashtable. Besides pointing to the
- * shared hashtable entry (PgStatShared_HashEntry) PgStat_EntryRef also
- * contains a pointer to the shared statistics data, as a process-local
- * address, to reduce access costs.
+ * To avoid contention on the shared stats hashtables, each backend has a
+ * backend-local hashtable (pgStatEntryRefHash) containing references
+ * (PgStat_EntryRef) to shared stats hashtable entries. The shared stats
+ * hashtables only need to be accessed when no prior reference is found in the
+ * local hashtable. Besides pointing to the shared stats hashtable entry
+ * (PgStatShared_HashEntry), PgStat_EntryRef also contains a process-local
+ * pointer to the shared statistics data, to reduce access costs.
  *
  * The names for structs stored in shared memory are prefixed with
  * PgStatShared instead of PgStat. Each stats entry in shared memory is
@@ -1199,52 +1198,57 @@ pgstat_build_snapshot(void)
 	/*
 	 * Snapshot all variable stats.
 	 */
-	dshash_seq_init(&hstat, pgStatLocal.shared_hash, false);
-	while ((p = dshash_seq_next(&hstat)) != NULL)
+	for (int h = 0; h < pgStatLocal.num_var_hashes; h++)
 	{
-		PgStat_Kind kind = p->key.kind;
-		const PgStat_KindInfo *kind_info = pgstat_get_kind_info(kind);
-		bool		found;
-		PgStat_SnapshotEntry *entry;
-		PgStatShared_Common *stats_data;
+		dshash_table *hash = pgStatLocal.var_hashes[h];
 
-		/*
-		 * Check if the stats object should be included in the snapshot.
-		 * Unless the stats kind can be accessed from all databases (e.g.,
-		 * database stats themselves), we only include stats for the current
-		 * database or objects not associated with a database (e.g. shared
-		 * relations).
-		 */
-		if (p->key.dboid != MyDatabaseId &&
-			p->key.dboid != InvalidOid &&
-			!kind_info->accessed_across_databases)
-			continue;
+		dshash_seq_init(&hstat, hash, false);
+		while ((p = dshash_seq_next(&hstat)) != NULL)
+		{
+			PgStat_Kind kind = p->key.kind;
+			const PgStat_KindInfo *kind_info = pgstat_get_kind_info(kind);
+			bool		found;
+			PgStat_SnapshotEntry *entry;
+			PgStatShared_Common *stats_data;
 
-		if (p->dropped)
-			continue;
+			/*
+			 * Check if the stats object should be included in the snapshot.
+			 * Unless the stats kind can be accessed from all databases (e.g.,
+			 * database stats themselves), we only include stats for the
+			 * current database or objects not associated with a database
+			 * (e.g. shared relations).
+			 */
+			if (p->key.dboid != MyDatabaseId &&
+				p->key.dboid != InvalidOid &&
+				!kind_info->accessed_across_databases)
+				continue;
 
-		Assert(pg_atomic_read_u32(&p->refcount) > 0);
+			if (p->dropped)
+				continue;
 
-		stats_data = dsa_get_address(pgStatLocal.dsa, p->body);
-		Assert(stats_data);
+			Assert(pg_atomic_read_u32(&p->refcount) > 0);
 
-		entry = pgstat_snapshot_insert(pgStatLocal.snapshot.stats, p->key, &found);
-		Assert(!found);
+			stats_data = dsa_get_address(pgStatLocal.kind_dsa[kind], p->body);
+			Assert(stats_data);
 
-		entry->data = MemoryContextAlloc(pgStatLocal.snapshot.context,
-										 pgstat_get_entry_len(kind));
+			entry = pgstat_snapshot_insert(pgStatLocal.snapshot.stats, p->key, &found);
+			Assert(!found);
 
-		/*
-		 * Acquire the LWLock directly instead of using
-		 * pg_stat_lock_entry_shared() which requires a reference.
-		 */
-		LWLockAcquire(&stats_data->lock, LW_SHARED);
-		memcpy(entry->data,
-			   pgstat_get_entry_data(kind, stats_data),
-			   pgstat_get_entry_len(kind));
-		LWLockRelease(&stats_data->lock);
+			entry->data = MemoryContextAlloc(pgStatLocal.snapshot.context,
+											 pgstat_get_entry_len(kind));
+
+			/*
+			 * Acquire the LWLock directly instead of using
+			 * pg_stat_lock_entry_shared() which requires a reference.
+			 */
+			LWLockAcquire(&stats_data->lock, LW_SHARED);
+			memcpy(entry->data,
+				   pgstat_get_entry_data(kind, stats_data),
+				   pgstat_get_entry_len(kind));
+			LWLockRelease(&stats_data->lock);
+		}
+		dshash_seq_term(&hstat);
 	}
-	dshash_seq_term(&hstat);
 
 	/*
 	 * Build snapshot of all fixed-numbered stats.
@@ -1584,6 +1588,10 @@ pgstat_register_kind(PgStat_Kind kind, const PgStat_KindInfo *kind_info)
 			ereport(ERROR,
 					(errmsg("failed to register custom cumulative statistics \"%s\" with ID %u", kind_info->name, kind),
 					 errhint("Custom cumulative statistics cannot use entry count tracking for fixed-numbered objects.")));
+		if (kind_info->own_hash)
+			ereport(ERROR,
+					(errmsg("failed to register custom cumulative statistics \"%s\" with ID %u", kind_info->name, kind),
+					 errhint("Custom cumulative statistics cannot use a dedicated hash table for fixed-numbered objects.")));
 	}
 	else
 	{
@@ -1737,82 +1745,88 @@ pgstat_write_statsfile(void)
 	}
 
 	/*
-	 * Walk through the stats entries
+	 * Walk through the stats entries, as long as writes can happen (note that
+	 * switching to a STATS_DISCARD status is possible while walking through).
 	 */
-	dshash_seq_init(&hstat, pgStatLocal.shared_hash, false);
-	while ((ps = dshash_seq_next(&hstat)) != NULL)
+	for (int h = 0; h < pgStatLocal.num_var_hashes && status == STATS_WRITE; h++)
 	{
-		PgStatShared_Common *shstats;
-		const PgStat_KindInfo *kind_info = NULL;
+		dshash_table *hash = pgStatLocal.var_hashes[h];
 
-		/*
-		 * We should not see any "dropped" entries when writing the stats
-		 * file, as all backends and auxiliary processes should have cleaned
-		 * up their references before they terminated.
-		 *
-		 * However, since we are already shutting down, it is not worth
-		 * crashing the server over any potential cleanup issues, so we simply
-		 * skip such entries if encountered.
-		 */
-		Assert(!ps->dropped);
-		if (ps->dropped)
-			continue;
-
-		/*
-		 * This discards data related to custom stats kinds that are unknown
-		 * to this process.
-		 */
-		if (!pgstat_is_kind_valid(ps->key.kind))
+		dshash_seq_init(&hstat, hash, false);
+		while ((ps = dshash_seq_next(&hstat)) != NULL)
 		{
-			elog(WARNING, "found unknown stats entry %u/%u/%" PRIu64,
-				 ps->key.kind, ps->key.dboid,
-				 ps->key.objid);
-			continue;
-		}
-
-		shstats = (PgStatShared_Common *) dsa_get_address(pgStatLocal.dsa, ps->body);
-
-		kind_info = pgstat_get_kind_info(ps->key.kind);
-
-		/* if not dropped the valid-entry refcount should exist */
-		Assert(pg_atomic_read_u32(&ps->refcount) > 0);
-
-		/* skip if no need to write to file */
-		if (!kind_info->write_to_file)
-			continue;
-
-		if (!kind_info->to_serialized_name)
-		{
-			/* normal stats entry, identified by PgStat_HashKey */
-			fputc(PGSTAT_FILE_ENTRY_HASH, fpout);
-			write_chunk_s(fpout, &ps->key);
-		}
-		else
-		{
-			/* stats entry identified by name on disk (e.g. slots) */
-			NameData	name;
-
-			kind_info->to_serialized_name(&ps->key, shstats, &name);
-
-			fputc(PGSTAT_FILE_ENTRY_NAME, fpout);
-			write_chunk_s(fpout, &ps->key.kind);
-			write_chunk_s(fpout, &name);
-		}
-
-		/* Write except the header part of the entry */
-		write_chunk(fpout,
-					pgstat_get_entry_data(ps->key.kind, shstats),
-					pgstat_get_entry_len(ps->key.kind));
-
-		/* Write more data for the entry, if required */
-		if (kind_info->to_serialized_data &&
-			!kind_info->to_serialized_data(&ps->key, shstats, fpout))
-		{
-			status = STATS_DISCARD;
-			break;
+			PgStatShared_Common *shstats;
+			const PgStat_KindInfo *kind_info = NULL;
+
+			/*
+			 * We should not see any "dropped" entries when writing the stats
+			 * file, as all backends and auxiliary processes should have
+			 * cleaned up their references before they terminated.
+			 *
+			 * However, since we are already shutting down, it is not worth
+			 * crashing the server over any potential cleanup issues, so we
+			 * simply skip such entries if encountered.
+			 */
+			Assert(!ps->dropped);
+			if (ps->dropped)
+				continue;
+
+			/*
+			 * This discards data related to custom stats kinds that are
+			 * unknown to this process.
+			 */
+			if (!pgstat_is_kind_valid(ps->key.kind))
+			{
+				elog(WARNING, "found unknown stats entry %u/%u/%" PRIu64,
+					 ps->key.kind, ps->key.dboid,
+					 ps->key.objid);
+				continue;
+			}
+
+			kind_info = pgstat_get_kind_info(ps->key.kind);
+			shstats = (PgStatShared_Common *)
+				dsa_get_address(pgStatLocal.kind_dsa[ps->key.kind], ps->body);
+
+			/* if not dropped the valid-entry refcount should exist */
+			Assert(pg_atomic_read_u32(&ps->refcount) > 0);
+
+			/* skip if no need to write to file */
+			if (!kind_info->write_to_file)
+				continue;
+
+			if (!kind_info->to_serialized_name)
+			{
+				/* normal stats entry, identified by PgStat_HashKey */
+				fputc(PGSTAT_FILE_ENTRY_HASH, fpout);
+				write_chunk_s(fpout, &ps->key);
+			}
+			else
+			{
+				/* stats entry identified by name on disk (e.g. slots) */
+				NameData	name;
+
+				kind_info->to_serialized_name(&ps->key, shstats, &name);
+
+				fputc(PGSTAT_FILE_ENTRY_NAME, fpout);
+				write_chunk_s(fpout, &ps->key.kind);
+				write_chunk_s(fpout, &name);
+			}
+
+			/* Write except the header part of the entry */
+			write_chunk(fpout,
+						pgstat_get_entry_data(ps->key.kind, shstats),
+						pgstat_get_entry_len(ps->key.kind));
+
+			/* Write more data for the entry, if required */
+			if (kind_info->to_serialized_data &&
+				!kind_info->to_serialized_data(&ps->key, shstats, fpout))
+			{
+				status = STATS_DISCARD;
+				break;
+			}
 		}
+		dshash_seq_term(&hstat);
 	}
-	dshash_seq_term(&hstat);
 
 	/*
 	 * No more output to be done. Close the temp file and replace the old
@@ -2110,12 +2124,12 @@ pgstat_read_statsfile(void)
 							 key.objid, t);
 					}
 
-					p = dshash_find_or_insert_extended(pgStatLocal.shared_hash,
+					p = dshash_find_or_insert_extended(pgStatLocal.kind_hash[key.kind],
 													   &key, &found,
 													   DSHASH_INSERT_NO_OOM);
 					if (!p)
 					{
-						dsa_free(pgStatLocal.dsa, chunk);
+						dsa_free(pgStatLocal.kind_dsa[key.kind], chunk);
 
 						/*
 						 * for the same reason as previously, ERROR not
@@ -2129,8 +2143,8 @@ pgstat_read_statsfile(void)
 					/* don't allow duplicate entries */
 					if (found)
 					{
-						dshash_release_lock(pgStatLocal.shared_hash, p);
-						dsa_free(pgStatLocal.dsa, chunk);
+						dshash_release_lock(pgStatLocal.kind_hash[key.kind], p);
+						dsa_free(pgStatLocal.kind_dsa[key.kind], chunk);
 						elog(WARNING, "found duplicate stats entry %u/%u/%" PRIu64 " of type %c",
 							 key.kind, key.dboid,
 							 key.objid, t);
@@ -2138,7 +2152,7 @@ pgstat_read_statsfile(void)
 					}
 
 					header = pgstat_init_entry(key.kind, p, chunk);
-					dshash_release_lock(pgStatLocal.shared_hash, p);
+					dshash_release_lock(pgStatLocal.kind_hash[key.kind], p);
 
 					if (!read_chunk(fpin,
 									pgstat_get_entry_data(key.kind, header),
diff --git a/src/backend/utils/activity/pgstat_shmem.c b/src/backend/utils/activity/pgstat_shmem.c
index 1858d68f918..9b571013dbd 100644
--- a/src/backend/utils/activity/pgstat_shmem.c
+++ b/src/backend/utils/activity/pgstat_shmem.c
@@ -66,7 +66,7 @@ const ShmemCallbacks StatsShmemCallbacks = {
 	.init_fn = StatsShmemInit,
 };
 
-/* parameter for the shared hash */
+/* Parameters for shared hashtables. */
 static const dshash_parameters dsh_params = {
 	sizeof(PgStat_HashKey),
 	sizeof(PgStatShared_HashEntry),
@@ -105,8 +105,8 @@ static MemoryContext pgStatEntryRefHashContext = NULL;
  */
 
 /*
- * The size of the shared memory allocation for stats stored in the shared
- * stats hash table. This allocation will be done as part of the main shared
+ * The size of the shared memory allocation for stats stored in a stats
+ * hashtable. This allocation will be done as part of the main shared
  * memory, rather than dynamic shared memory, allowing it to be initialized in
  * postmaster.
  */
@@ -139,6 +139,17 @@ StatsShmemSize(void)
 	sz = MAXALIGN(sizeof(PgStat_ShmemControl));
 	sz = add_size(sz, pgstat_dsa_init_size());
 
+	/* Add per-kind DSA space for variable-numbered own_hash kinds. */
+	for (PgStat_Kind kind = PGSTAT_KIND_MIN; kind <= PGSTAT_KIND_MAX; kind++)
+	{
+		const PgStat_KindInfo *kind_info = pgstat_get_kind_info(kind);
+
+		if (!kind_info || kind_info->fixed_amount || !kind_info->own_hash)
+			continue;
+
+		sz = add_size(sz, pgstat_dsa_init_size());
+	}
+
 	/* Add shared memory for all the custom fixed-numbered statistics */
 	for (PgStat_Kind kind = PGSTAT_KIND_CUSTOM_MIN; kind <= PGSTAT_KIND_CUSTOM_MAX; kind++)
 	{
@@ -210,6 +221,41 @@ StatsShmemInit(void *arg)
 	/* lift limit set above */
 	dsa_set_size_limit(dsa, -1);
 
+	/*
+	 * Create dedicated DSA areas and hash tables for kinds that have
+	 * variable-numbered stats and set own_hash to true.
+	 */
+	memset(ctl->raw_kind_dsa_area, 0, sizeof(ctl->raw_kind_dsa_area));
+	for (PgStat_Kind kind = PGSTAT_KIND_MIN; kind <= PGSTAT_KIND_MAX; kind++)
+	{
+		const PgStat_KindInfo *kind_info = pgstat_get_kind_info(kind);
+		dsa_area   *kind_dsa;
+		dshash_table *kind_dsh;
+
+		if (!kind_info || kind_info->fixed_amount || !kind_info->own_hash)
+			continue;
+
+		/* Create a per-kind DSA in pre-allocated shared memory */
+		ctl->raw_kind_dsa_area[kind] = p;
+		p += pgstat_dsa_init_size();
+		kind_dsa = dsa_create_in_place(ctl->raw_kind_dsa_area[kind],
+									   pgstat_dsa_init_size(),
+									   LWTRANCHE_PGSTATS_DSA, NULL);
+		dsa_pin(kind_dsa);
+
+		/* Temporarily limit to keep dshash in the in-place area */
+		dsa_set_size_limit(kind_dsa, pgstat_dsa_init_size());
+
+		kind_dsh = dshash_create(kind_dsa, &dsh_params, NULL);
+		ctl->kind_hash_handles[kind] = dshash_get_hash_table_handle(kind_dsh);
+
+		dsa_set_size_limit(kind_dsa, -1);
+
+		/* detach, as postmaster never accesses them again. */
+		dshash_detach(kind_dsh);
+		dsa_detach(kind_dsa);
+	}
+
 	/*
 	 * Postmaster will never access these again, thus free the local
 	 * dsa/dshash references.
@@ -255,20 +301,61 @@ StatsShmemInit(void *arg)
 void
 pgstat_attach_shmem(void)
 {
+	dsa_area   *shared_dsa;
+	dshash_table *shared_hash;
 	MemoryContext oldcontext;
 
-	Assert(pgStatLocal.dsa == NULL);
+	Assert(pgStatLocal.num_var_hashes == 0);
 
 	/* stats shared memory persists for the backend lifetime */
 	oldcontext = MemoryContextSwitchTo(TopMemoryContext);
 
-	pgStatLocal.dsa = dsa_attach_in_place(pgStatLocal.shmem->raw_dsa_area,
-										  NULL);
-	dsa_pin_mapping(pgStatLocal.dsa);
+	shared_dsa = dsa_attach_in_place(pgStatLocal.shmem->raw_dsa_area,
+									 NULL);
+	dsa_pin_mapping(shared_dsa);
 
-	pgStatLocal.shared_hash = dshash_attach(pgStatLocal.dsa, &dsh_params,
-											pgStatLocal.shmem->hash_handle,
-											NULL);
+	shared_hash = dshash_attach(shared_dsa, &dsh_params,
+								pgStatLocal.shmem->hash_handle,
+								NULL);
+
+	/* Build per-kind mappings for the hash tables. */
+	pgStatLocal.var_hashes[pgStatLocal.num_var_hashes++] = shared_hash;
+
+	for (PgStat_Kind kind = PGSTAT_KIND_MIN; kind <= PGSTAT_KIND_MAX; kind++)
+	{
+		const PgStat_KindInfo *kind_info = pgstat_get_kind_info(kind);
+
+		if (pgStatLocal.shmem->raw_kind_dsa_area[kind] != NULL)
+		{
+			dsa_area   *kind_dsa;
+
+			/* per-kind hashtable */
+			kind_dsa =
+				dsa_attach_in_place(pgStatLocal.shmem->raw_kind_dsa_area[kind],
+									NULL);
+			dsa_pin_mapping(kind_dsa);
+
+			pgStatLocal.kind_hash[kind] =
+				dshash_attach(kind_dsa, &dsh_params,
+							  pgStatLocal.shmem->kind_hash_handles[kind],
+							  NULL);
+			pgStatLocal.kind_dsa[kind] = kind_dsa;
+			pgStatLocal.var_hashes[pgStatLocal.num_var_hashes++] =
+				pgStatLocal.kind_hash[kind];
+		}
+		else if (kind_info && !kind_info->fixed_amount)
+		{
+			/* main shared hashtable */
+			pgStatLocal.kind_hash[kind] = shared_hash;
+			pgStatLocal.kind_dsa[kind] = shared_dsa;
+		}
+		else
+		{
+			/* unassigned kind ID */
+			pgStatLocal.kind_hash[kind] = NULL;
+			pgStatLocal.kind_dsa[kind] = NULL;
+		}
+	}
 
 	MemoryContextSwitchTo(oldcontext);
 }
@@ -276,15 +363,28 @@ pgstat_attach_shmem(void)
 void
 pgstat_detach_shmem(void)
 {
-	Assert(pgStatLocal.dsa);
+	Assert(pgStatLocal.num_var_hashes > 0);
 
 	/* we shouldn't leave references to shared stats */
 	pgstat_release_all_entry_refs(false);
 
-	dshash_detach(pgStatLocal.shared_hash);
-	pgStatLocal.shared_hash = NULL;
+	for (int h = 0; h < pgStatLocal.num_var_hashes; h++)
+	{
+		dsa_area   *dsa = dshash_get_dsa_area(pgStatLocal.var_hashes[h]);
 
-	dsa_detach(pgStatLocal.dsa);
+		dshash_detach(pgStatLocal.var_hashes[h]);
+		dsa_detach(dsa);
+		pgStatLocal.var_hashes[h] = NULL;
+	}
+
+	pgStatLocal.num_var_hashes = 0;
+
+	/* Reset per-kind mappings. */
+	for (PgStat_Kind kind = PGSTAT_KIND_MIN; kind <= PGSTAT_KIND_MAX; kind++)
+	{
+		pgStatLocal.kind_hash[kind] = NULL;
+		pgStatLocal.kind_dsa[kind] = NULL;
+	}
 
 	/*
 	 * dsa_detach() does not decrement the DSA reference count as no segment
@@ -293,7 +393,11 @@ pgstat_detach_shmem(void)
 	 */
 	dsa_release_in_place(pgStatLocal.shmem->raw_dsa_area);
 
-	pgStatLocal.dsa = NULL;
+	for (PgStat_Kind kind = PGSTAT_KIND_MIN; kind <= PGSTAT_KIND_MAX; kind++)
+	{
+		if (pgStatLocal.shmem->raw_kind_dsa_area[kind] != NULL)
+			dsa_release_in_place(pgStatLocal.shmem->raw_kind_dsa_area[kind]);
+	}
 }
 
 
@@ -312,7 +416,7 @@ pgstat_alloc_entry_body(PgStat_Kind kind)
 {
 	const PgStat_KindInfo *kind_info = pgstat_get_kind_info(kind);
 
-	return dsa_allocate_extended(pgStatLocal.dsa,
+	return dsa_allocate_extended(pgStatLocal.kind_dsa[kind],
 								 kind_info->shared_size,
 								 DSA_ALLOC_ZERO | DSA_ALLOC_NO_OOM);
 }
@@ -347,7 +451,7 @@ pgstat_init_entry(PgStat_Kind kind,
 	pg_atomic_init_u32(&shhashent->generation, 0);
 	shhashent->dropped = false;
 
-	shheader = dsa_get_address(pgStatLocal.dsa, chunk);
+	shheader = dsa_get_address(pgStatLocal.kind_dsa[kind], chunk);
 	shheader->magic = 0xdeadbeef;
 
 	/* Link the new entry from the hash entry. */
@@ -367,7 +471,8 @@ pgstat_reinit_entry(PgStat_Kind kind, PgStatShared_HashEntry *shhashent)
 {
 	PgStatShared_Common *shheader;
 
-	shheader = dsa_get_address(pgStatLocal.dsa, shhashent->body);
+	shheader = dsa_get_address(pgStatLocal.kind_dsa[kind],
+							   shhashent->body);
 
 	/* mark as not dropped anymore */
 	pg_atomic_fetch_add_u32(&shhashent->refcount, 1);
@@ -405,6 +510,7 @@ pgstat_setup_shared_refs(void)
  */
 static void
 pgstat_acquire_entry_ref(PgStat_EntryRef *entry_ref,
+						 dshash_table *hash,
 						 PgStatShared_HashEntry *shhashent,
 						 PgStatShared_Common *shheader)
 {
@@ -422,7 +528,7 @@ pgstat_acquire_entry_ref(PgStat_EntryRef *entry_ref,
 	 * LWLock can process a pending interrupt, and callers may catch the
 	 * resulting error and continue using the backend-local cache.
 	 */
-	dshash_release_lock(pgStatLocal.shared_hash, shhashent);
+	dshash_release_lock(hash, shhashent);
 }
 
 /*
@@ -506,6 +612,8 @@ pgstat_get_entry_ref(PgStat_Kind kind, Oid dboid, uint64 objid, bool create,
 	PgStatShared_HashEntry *shhashent;
 	PgStatShared_Common *shheader = NULL;
 	PgStat_EntryRef *entry_ref;
+	dshash_table *hash;
+	dsa_area   *dsa;
 
 	key.kind = kind;
 	key.dboid = dboid;
@@ -517,7 +625,7 @@ pgstat_get_entry_ref(PgStat_Kind kind, Oid dboid, uint64 objid, bool create,
 	 */
 	Assert(create || created_entry == NULL);
 	pgstat_assert_is_up();
-	Assert(pgStatLocal.shared_hash != NULL);
+	Assert(pgStatLocal.num_var_hashes > 0);
 	Assert(!pgStatLocal.shmem->is_shutdown);
 
 	pgstat_setup_memcxt();
@@ -545,11 +653,15 @@ pgstat_get_entry_ref(PgStat_Kind kind, Oid dboid, uint64 objid, bool create,
 
 	Assert(entry_ref != NULL);
 
+	/* Per-kind hashtable and DSA area */
+	hash = pgStatLocal.kind_hash[kind];
+	dsa = pgStatLocal.kind_dsa[kind];
+
 	/*
 	 * Do a lookup in the hash table first - it's quite likely that the entry
 	 * already exists, and that way we only need a shared lock.
 	 */
-	shhashent = dshash_find(pgStatLocal.shared_hash, &key, false);
+	shhashent = dshash_find(hash, &key, false);
 
 	if (create && !shhashent)
 	{
@@ -573,16 +685,16 @@ pgstat_get_entry_ref(PgStat_Kind kind, Oid dboid, uint64 objid, bool create,
 		 * lookup. If so, fall through to the same path as if we'd have if it
 		 * already had been created before the dshash_find() calls.
 		 */
-		shhashent = dshash_find_or_insert_extended(pgStatLocal.shared_hash,
+		shhashent = dshash_find_or_insert_extended(hash,
 												   &key, &shfound,
 												   DSHASH_INSERT_NO_OOM);
 		if (!shhashent)
 		{
-			dsa_free(pgStatLocal.dsa, chunk);
+			dsa_free(dsa, chunk);
 
 			/*
-			 * Clean up the local reference when failing insert into the
-			 * shared hashtable.
+			 * Clean up the local reference when failing insert into the stats
+			 * hashtable.
 			 */
 			pgstat_release_entry_ref(key, entry_ref, false);
 			ereport(ERROR,
@@ -595,7 +707,7 @@ pgstat_get_entry_ref(PgStat_Kind kind, Oid dboid, uint64 objid, bool create,
 		if (!shfound)
 		{
 			shheader = pgstat_init_entry(kind, shhashent, chunk);
-			pgstat_acquire_entry_ref(entry_ref, shhashent, shheader);
+			pgstat_acquire_entry_ref(entry_ref, hash, shhashent, shheader);
 
 			if (created_entry != NULL)
 				*created_entry = true;
@@ -604,7 +716,7 @@ pgstat_get_entry_ref(PgStat_Kind kind, Oid dboid, uint64 objid, bool create,
 		}
 
 		/* Concurrent insert won; drop the unused body. */
-		dsa_free(pgStatLocal.dsa, chunk);
+		dsa_free(dsa, chunk);
 	}
 
 	if (!shhashent)
@@ -637,7 +749,7 @@ pgstat_get_entry_ref(PgStat_Kind kind, Oid dboid, uint64 objid, bool create,
 			 * in the shared entry for any remaining local references.
 			 */
 			shheader = pgstat_reinit_entry(kind, shhashent);
-			pgstat_acquire_entry_ref(entry_ref, shhashent, shheader);
+			pgstat_acquire_entry_ref(entry_ref, hash, shhashent, shheader);
 
 			if (created_entry != NULL)
 				*created_entry = true;
@@ -646,15 +758,15 @@ pgstat_get_entry_ref(PgStat_Kind kind, Oid dboid, uint64 objid, bool create,
 		}
 		else if (shhashent->dropped)
 		{
-			dshash_release_lock(pgStatLocal.shared_hash, shhashent);
+			dshash_release_lock(hash, shhashent);
 			pgstat_release_entry_ref(key, entry_ref, false);
 
 			return NULL;
 		}
 		else
 		{
-			shheader = dsa_get_address(pgStatLocal.dsa, shhashent->body);
-			pgstat_acquire_entry_ref(entry_ref, shhashent, shheader);
+			shheader = dsa_get_address(dsa, shhashent->body);
+			pgstat_acquire_entry_ref(entry_ref, hash, shhashent, shheader);
 
 			return entry_ref;
 		}
@@ -695,9 +807,8 @@ pgstat_release_entry_ref(PgStat_HashKey key, PgStat_EntryRef *entry_ref,
 			/* only dropped entries can reach a 0 refcount */
 			Assert(entry_ref->shared_entry->dropped);
 
-			shent = dshash_find(pgStatLocal.shared_hash,
-								&entry_ref->shared_entry->key,
-								true);
+			shent = dshash_find(pgStatLocal.kind_hash[key.kind],
+								&entry_ref->shared_entry->key, true);
 			if (!shent)
 				elog(ERROR, "could not find just referenced shared stats entry");
 
@@ -720,7 +831,7 @@ pgstat_release_entry_ref(PgStat_HashKey key, PgStat_EntryRef *entry_ref,
 				 * Shared stats entry has been reinitialized, so do not drop
 				 * its shared entry, only release its lock.
 				 */
-				dshash_release_lock(pgStatLocal.shared_hash, shent);
+				dshash_release_lock(pgStatLocal.kind_hash[key.kind], shent);
 			}
 		}
 	}
@@ -943,11 +1054,11 @@ pgstat_free_entry(PgStatShared_HashEntry *shent, dshash_seq_status *hstat)
 	pdsa = shent->body;
 
 	if (!hstat)
-		dshash_delete_entry(pgStatLocal.shared_hash, shent);
+		dshash_delete_entry(pgStatLocal.kind_hash[kind], shent);
 	else
 		dshash_delete_current(hstat);
 
-	dsa_free(pgStatLocal.dsa, pdsa);
+	dsa_free(pgStatLocal.kind_dsa[kind], pdsa);
 
 	/* Decrement entry count, if required. */
 	if (pgstat_get_kind_info(kind)->track_entry_count)
@@ -986,7 +1097,7 @@ pgstat_drop_entry_internal(PgStatShared_HashEntry *shent,
 	else
 	{
 		if (!hstat)
-			dshash_release_lock(pgStatLocal.shared_hash, shent);
+			dshash_release_lock(pgStatLocal.kind_hash[shent->key.kind], shent);
 		return false;
 	}
 }
@@ -1003,7 +1114,7 @@ pgstat_drop_database_and_contents(Oid dboid)
 
 	Assert(OidIsValid(dboid));
 
-	Assert(pgStatLocal.shared_hash != NULL);
+	Assert(pgStatLocal.num_var_hashes > 0);
 
 	/*
 	 * This backend might very well be the only backend holding a reference to
@@ -1011,30 +1122,34 @@ pgstat_drop_database_and_contents(Oid dboid)
 	 * being cleaned up till later.
 	 *
 	 * Doing this separately from the dshash iteration below avoids having to
-	 * do so while holding a partition lock on the shared hashtable.
+	 * do so while holding a partition lock on any of the shared hashtables.
 	 */
 	pgstat_release_db_entry_refs(dboid);
 
-	/* some of the dshash entries are to be removed, take exclusive lock. */
-	dshash_seq_init(&hstat, pgStatLocal.shared_hash, true);
-	while ((p = dshash_seq_next(&hstat)) != NULL)
+	for (int h = 0; h < pgStatLocal.num_var_hashes; h++)
 	{
-		if (p->dropped)
-			continue;
-
-		if (p->key.dboid != dboid)
-			continue;
-
-		if (!pgstat_drop_entry_internal(p, &hstat))
+		/* some of the dshash entries are to be removed, take exclusive lock. */
+		dshash_seq_init(&hstat, pgStatLocal.var_hashes[h], true);
+		while ((p = dshash_seq_next(&hstat)) != NULL)
 		{
-			/*
-			 * Even statistics for a dropped database might currently be
-			 * accessed (consider e.g. database stats for pg_stat_database).
-			 */
-			not_freed_count++;
+			if (p->dropped)
+				continue;
+
+			if (p->key.dboid != dboid)
+				continue;
+
+			if (!pgstat_drop_entry_internal(p, &hstat))
+			{
+				/*
+				 * Even statistics for a dropped database might currently be
+				 * accessed (consider e.g. database stats for
+				 * pg_stat_database).
+				 */
+				not_freed_count++;
+			}
 		}
+		dshash_seq_term(&hstat);
 	}
-	dshash_seq_term(&hstat);
 
 	/*
 	 * If some of the stats data could not be freed, signal the reference
@@ -1065,6 +1180,18 @@ pgstat_drop_entry(PgStat_Kind kind, Oid dboid, uint64 objid,
 	PgStatShared_HashEntry *shent;
 	bool		freed = true;
 
+	Assert(kind >= PGSTAT_KIND_MIN && kind <= PGSTAT_KIND_MAX);
+
+	/* Check case of unassigned kind ID */
+	if (pgStatLocal.kind_hash[kind] == NULL)
+	{
+		if (missing_ok)
+			return true;
+
+		elog(ERROR, "invalid stats kind for entry %u/%u/%" PRIu64,
+			 kind, dboid, objid);
+	}
+
 	key.kind = kind;
 	key.dboid = dboid;
 	key.objid = objid;
@@ -1080,8 +1207,8 @@ pgstat_drop_entry(PgStat_Kind kind, Oid dboid, uint64 objid,
 									 true);
 	}
 
-	/* mark entry in shared hashtable as deleted, drop if possible */
-	shent = dshash_find(pgStatLocal.shared_hash, &key, true);
+	/* mark entry in the stats hashtable as deleted, drop if possible */
+	shent = dshash_find(pgStatLocal.kind_hash[kind], &key, true);
 	if (shent)
 	{
 		if (shent->dropped)
@@ -1094,7 +1221,7 @@ pgstat_drop_entry(PgStat_Kind kind, Oid dboid, uint64 objid,
 					 shent->key.objid,
 					 pg_atomic_read_u32(&shent->refcount),
 					 pg_atomic_read_u32(&shent->generation));
-			dshash_release_lock(pgStatLocal.shared_hash, shent);
+			dshash_release_lock(pgStatLocal.kind_hash[kind], shent);
 			return true;
 		}
 
@@ -1114,8 +1241,8 @@ pgstat_drop_entry(PgStat_Kind kind, Oid dboid, uint64 objid,
 }
 
 /*
- * Scan through the shared hashtable of stats, dropping statistics if
- * approved by the optional do_drop() function.
+ * Scan through the stats hashtables, dropping statistics if approved by the
+ * optional do_drop() function.
  */
 void
 pgstat_drop_matching_entries(bool (*do_drop) (PgStatShared_HashEntry *, Datum),
@@ -1125,38 +1252,41 @@ pgstat_drop_matching_entries(bool (*do_drop) (PgStatShared_HashEntry *, Datum),
 	PgStatShared_HashEntry *ps;
 	uint64		not_freed_count = 0;
 
-	/* entries are removed, take an exclusive lock */
-	dshash_seq_init(&hstat, pgStatLocal.shared_hash, true);
-	while ((ps = dshash_seq_next(&hstat)) != NULL)
+	for (int h = 0; h < pgStatLocal.num_var_hashes; h++)
 	{
-		if (ps->dropped)
-			continue;
-
-		if (do_drop != NULL && !do_drop(ps, match_data))
-			continue;
-
-		/* delete local reference */
-		if (pgStatEntryRefHash)
+		/* entries are removed, take an exclusive lock */
+		dshash_seq_init(&hstat, pgStatLocal.var_hashes[h], true);
+		while ((ps = dshash_seq_next(&hstat)) != NULL)
 		{
-			PgStat_EntryRefHashEntry *lohashent =
-				pgstat_entry_ref_hash_lookup(pgStatEntryRefHash, ps->key);
+			if (ps->dropped)
+				continue;
 
-			if (lohashent)
-				pgstat_release_entry_ref(lohashent->key, lohashent->entry_ref,
-										 true);
+			if (do_drop != NULL && !do_drop(ps, match_data))
+				continue;
+
+			/* delete local reference */
+			if (pgStatEntryRefHash)
+			{
+				PgStat_EntryRefHashEntry *lohashent =
+					pgstat_entry_ref_hash_lookup(pgStatEntryRefHash, ps->key);
+
+				if (lohashent)
+					pgstat_release_entry_ref(lohashent->key, lohashent->entry_ref,
+											 true);
+			}
+
+			if (!pgstat_drop_entry_internal(ps, &hstat))
+				not_freed_count++;
 		}
-
-		if (!pgstat_drop_entry_internal(ps, &hstat))
-			not_freed_count++;
+		dshash_seq_term(&hstat);
 	}
-	dshash_seq_term(&hstat);
 
 	if (not_freed_count > 0)
 		pgstat_request_entry_refs_gc();
 }
 
 /*
- * Scan through the shared hashtable of stats and drop all entries.
+ * Scan through the stats hashtables and drop all entries.
  */
 void
 pgstat_drop_all_entries(void)
@@ -1177,6 +1307,38 @@ shared_stat_reset_contents(PgStat_Kind kind, PgStatShared_Common *header,
 		kind_info->reset_timestamp_cb(header, ts);
 }
 
+static void
+pgstat_reset_matching_entries_in_hash(dshash_table *hash,
+									  bool (*do_reset) (PgStatShared_HashEntry *, Datum),
+									  Datum match_data,
+									  TimestampTz ts)
+{
+	dshash_seq_status hstat;
+	PgStatShared_HashEntry *p;
+
+	/* dshash entry is not modified, take shared lock */
+	dshash_seq_init(&hstat, hash, false);
+	while ((p = dshash_seq_next(&hstat)) != NULL)
+	{
+		PgStatShared_Common *header;
+
+		if (p->dropped)
+			continue;
+
+		if (!do_reset(p, match_data))
+			continue;
+
+		header = dsa_get_address(pgStatLocal.kind_dsa[p->key.kind], p->body);
+
+		LWLockAcquire(&header->lock, LW_EXCLUSIVE);
+
+		shared_stat_reset_contents(p->key.kind, header, ts);
+
+		LWLockRelease(&header->lock);
+	}
+	dshash_seq_term(&hstat);
+}
+
 /*
  * Reset one variable-numbered stats entry.
  */
@@ -1197,37 +1359,20 @@ pgstat_reset_entry(PgStat_Kind kind, Oid dboid, uint64 objid, TimestampTz ts)
 }
 
 /*
- * Scan through the shared hashtable of stats, resetting statistics if
- * approved by the provided do_reset() function.
+ * Scan through the stats hashtables, resetting statistics if approved by the
+ * provided do_reset() function.
  */
 void
 pgstat_reset_matching_entries(bool (*do_reset) (PgStatShared_HashEntry *, Datum),
 							  Datum match_data, TimestampTz ts)
 {
-	dshash_seq_status hstat;
-	PgStatShared_HashEntry *p;
-
-	/* dshash entry is not modified, take shared lock */
-	dshash_seq_init(&hstat, pgStatLocal.shared_hash, false);
-	while ((p = dshash_seq_next(&hstat)) != NULL)
+	for (int h = 0; h < pgStatLocal.num_var_hashes; h++)
 	{
-		PgStatShared_Common *header;
-
-		if (p->dropped)
-			continue;
-
-		if (!do_reset(p, match_data))
-			continue;
-
-		header = dsa_get_address(pgStatLocal.dsa, p->body);
-
-		LWLockAcquire(&header->lock, LW_EXCLUSIVE);
-
-		shared_stat_reset_contents(p->key.kind, header, ts);
-
-		LWLockRelease(&header->lock);
+		pgstat_reset_matching_entries_in_hash(pgStatLocal.var_hashes[h],
+											  do_reset,
+											  match_data,
+											  ts);
 	}
-	dshash_seq_term(&hstat);
 }
 
 static bool
@@ -1239,7 +1384,12 @@ match_kind(PgStatShared_HashEntry *p, Datum match_data)
 void
 pgstat_reset_entries_of_kind(PgStat_Kind kind, TimestampTz ts)
 {
-	pgstat_reset_matching_entries(match_kind, Int32GetDatum(kind), ts);
+	Assert(pgStatLocal.kind_hash[kind] != NULL);
+
+	pgstat_reset_matching_entries_in_hash(pgStatLocal.kind_hash[kind],
+										  match_kind,
+										  Int32GetDatum(kind),
+										  ts);
 }
 
 static void
diff --git a/src/include/utils/pgstat_internal.h b/src/include/utils/pgstat_internal.h
index 201e57279b1..21403516e28 100644
--- a/src/include/utils/pgstat_internal.h
+++ b/src/include/utils/pgstat_internal.h
@@ -25,10 +25,10 @@
 /*
  * Types related to shared memory storage of statistics.
  *
- * Per-object statistics are stored in the "shared stats" hashtable. That
- * table's entries (PgStatShared_HashEntry) contain a pointer to the actual stats
- * data for the object (the size of the stats data varies depending on the
- * kind of stats). The table is keyed by PgStat_HashKey.
+ * Per-object statistics are stored in shared stats hashtables. Those tables'
+ * entries (PgStatShared_HashEntry) contain a pointer to the actual stats data
+ * for the object (the size of the stats data varies depending on the kind of
+ * stats). The tables are keyed by PgStat_HashKey.
  *
  * Once a backend has a reference to a shared stats entry, it increments the
  * entry's refcount. Even after stats data is dropped (e.g., due to a DROP
@@ -36,16 +36,16 @@
  * released.
  *
  * These refcounts, in combination with a backend local hashtable
- * (pgStatEntryRefHash, with entries pointing to PgStat_EntryRef) in front of
- * the shared hash table, mean that most stats work can happen without
- * touching the shared hash table, reducing contention.
+ * (pgStatEntryRefHash, with entries pointing to PgStat_EntryRef) that caches
+ * references to shared stats hashtable entries, mean that most stats work can
+ * happen without touching the shared stats hashtables, reducing contention.
  *
  * Once there are pending stats updates for a table PgStat_EntryRef->pending
  * is allocated to contain a working space for as-of-yet-unapplied stats
  * updates. Once the stats are flushed, PgStat_EntryRef->pending is freed.
  *
- * Each stat kind in the shared hash table has a fixed member
- * PgStatShared_Common as the first element.
+ * Each variable-numbered shared stats struct has PgStatShared_Common as the
+ * first element.
  */
 
 /*
@@ -186,7 +186,7 @@ typedef struct PgStat_EntryRef
 	 * stats eventually. Each stats kind utilizing pending data defines what
 	 * format its pending data has and needs to provide a
 	 * PgStat_KindInfo->flush_pending_cb callback to merge pending entries
-	 * into the shared stats hash table.
+	 * into shared stats hashtable entries.
 	 */
 	void	   *pending;
 	dlist_node	pending_node;	/* membership in pgStatPending list */
@@ -252,9 +252,19 @@ typedef struct PgStat_KindInfo
 	bool		track_entry_count:1;
 
 	/*
-	 * The size of an entry in the shared stats hash table (pointed to by
-	 * PgStatShared_HashEntry->body).  For fixed-numbered statistics, this is
-	 * the size of an entry in PgStat_ShmemControl->custom_data.
+	 * Should entries of this kind be stored in a dedicated dshash table
+	 * rather than the shared hash table? For variable-numbered stats only.
+	 *
+	 * A kind setting this option creates its own DSA area in the main shared
+	 * memory.  This is worth only for kinds expected to hold a large number
+	 * of entries.
+	 */
+	bool		own_hash:1;
+
+	/*
+	 * The size of an entry pointed to by PgStatShared_HashEntry->body.  For
+	 * fixed-numbered statistics, this is the size of an entry in
+	 * PgStat_ShmemControl->custom_data.
 	 */
 	uint32		shared_size;
 
@@ -541,7 +551,7 @@ typedef struct PgStatShared_Backend
 /*
  * Central shared memory entry for the cumulative stats system.
  *
- * Fixed amount stats, the dynamic shared memory hash table for
+ * Fixed amount stats, the dynamic shared memory hashtables for
  * non-fixed-amount stats, as well as remaining bits and pieces are all
  * reached from here.
  */
@@ -549,11 +559,12 @@ typedef struct PgStat_ShmemControl
 {
 	void	   *raw_dsa_area;
 
-	/*
-	 * Stats for variable-numbered objects are kept in this shared hash table.
-	 * See comment above PgStat_Kind for details.
-	 */
-	dshash_table_handle hash_handle;	/* shared dbstat hash */
+	/* Shared hash for variable-numbered stats kinds without own_hash set. */
+	dshash_table_handle hash_handle;	/* shared stats hash */
+
+	/* Dedicated hash and DSA for variable-numbered kinds with own_hash set. */
+	dshash_table_handle kind_hash_handles[PGSTAT_KIND_MAX + 1];
+	void	   *raw_kind_dsa_area[PGSTAT_KIND_MAX + 1];
 
 	/* Has the stats system already been shut down? Just a debugging check. */
 	bool		is_shutdown;
@@ -568,12 +579,12 @@ typedef struct PgStat_ShmemControl
 	pg_atomic_uint64 gc_request_count;
 
 	/*
-	 * Counters for the number of entries associated to a single stats kind
-	 * that uses variable-numbered objects stored in the shared hash table.
-	 * These counters can be enabled on a per-kind basis, when
-	 * track_entry_count is set.  This counter is incremented each time a new
-	 * entry is created (not reused) in the shared hash table, and is
-	 * decremented each time an entry is freed from the shared hash table.
+	 * Counters for the number of entries associated to a single
+	 * variable-numbered stats kind.  These counters can be enabled on a
+	 * per-kind basis, when track_entry_count is set.  A counter is
+	 * incremented each time a new entry is created (not reused) in a stats
+	 * hashtable, and is decremented each time an entry is freed from a stats
+	 * hashtable.
 	 */
 	pg_atomic_uint64 entry_counts[PGSTAT_KIND_MAX];
 
@@ -643,8 +654,12 @@ typedef struct PgStat_Snapshot
 typedef struct PgStat_LocalState
 {
 	PgStat_ShmemControl *shmem;
-	dsa_area   *dsa;
-	dshash_table *shared_hash;
+	dsa_area   *kind_dsa[PGSTAT_KIND_MAX + 1];
+	dshash_table *kind_hash[PGSTAT_KIND_MAX + 1];
+
+	/* Variable-numbered stats hash tables to iterate, built at attach time. */
+	dshash_table *var_hashes[PGSTAT_KIND_MAX + 1];
+	int			num_var_hashes;
 
 	/* the current statistics snapshot */
 	PgStat_Snapshot snapshot;
diff --git a/src/test/modules/test_custom_stats/t/001_custom_stats.pl b/src/test/modules/test_custom_stats/t/001_custom_stats.pl
index 69f2284229e..67e0c4e63d0 100644
--- a/src/test/modules/test_custom_stats/t/001_custom_stats.pl
+++ b/src/test/modules/test_custom_stats/t/001_custom_stats.pl
@@ -8,6 +8,7 @@
 # - Persistence across restarts.
 # - Loss after crash recovery.
 # - Resets for fixed-sized stats.
+# - Custom hash tables for variable-sized stats
 
 use strict;
 use warnings FATAL => 'all';
@@ -17,6 +18,7 @@ use PostgreSQL::Test::Utils;
 use Test::More;
 use File::Copy;
 
+my $result;
 my $node = PostgreSQL::Test::Cluster->new('main');
 $node->init;
 $node->append_conf('postgresql.conf',
@@ -28,7 +30,7 @@ $node->safe_psql('postgres', q(CREATE EXTENSION test_custom_var_stats));
 $node->safe_psql('postgres', q(CREATE EXTENSION test_custom_fixed_stats));
 
 # Verify custom stats kinds appear in pg_stat_kind_info.
-my $result = $node->safe_psql(
+$result = $node->safe_psql(
 	'postgres',
 	q(SELECT id, name, builtin, fixed_amount, accessed_across_databases,
 	         write_to_file
@@ -39,113 +41,155 @@ is( $result,
 26|test_custom_fixed_stats|f|t|f|t},
 	"custom stats kinds visible in pg_stat_kind_info");
 
-# Create entries for variable-sized stats.
-$node->safe_psql('postgres',
-	q(select test_custom_stats_var_create('entry1', 'Test entry 1')));
-$node->safe_psql('postgres',
-	q(select test_custom_stats_var_create('entry2', 'Test entry 2')));
-$node->safe_psql('postgres',
-	q(select test_custom_stats_var_create('entry3', 'Test entry 3')));
-$node->safe_psql('postgres',
-	q(select test_custom_stats_var_create('entry4', 'Test entry 4')));
+my @var_stats_kinds = (
+	{
+		name => 'test_custom_var_stats_own_hash',
+		own_hash => 't',
+	},
+	{
+		name => 'test_custom_var_stats',
+		own_hash => 'f',
+	});
 
-# Update counters: entry1=2, entry2=3, entry3=2, entry4=3, fixed=3
-$node->safe_psql('postgres',
-	q(select test_custom_stats_var_update('entry1')));
-$node->safe_psql('postgres',
-	q(select test_custom_stats_var_update('entry1')));
-$node->safe_psql('postgres',
-	q(select test_custom_stats_var_update('entry2')));
-$node->safe_psql('postgres',
-	q(select test_custom_stats_var_update('entry2')));
-$node->safe_psql('postgres',
-	q(select test_custom_stats_var_update('entry2')));
-$node->safe_psql('postgres',
-	q(select test_custom_stats_var_update('entry3')));
-$node->safe_psql('postgres',
-	q(select test_custom_stats_var_update('entry3')));
-$node->safe_psql('postgres',
-	q(select test_custom_stats_var_update('entry4')));
-$node->safe_psql('postgres',
-	q(select test_custom_stats_var_update('entry4')));
-$node->safe_psql('postgres',
-	q(select test_custom_stats_var_update('entry4')));
+# Initialize, to adjust later.
+$node->append_conf('postgresql.conf',
+	"test_custom_var_stats.kind = 'test_custom_var_stats'");
+
+# Test variable-sized stats with both dshash configurations.
+foreach my $stats_kind (@var_stats_kinds)
+{
+	my $kind_name = $stats_kind->{name};
+	my $own_hash = $stats_kind->{own_hash};
+
+	# Restart with the appropriate setting.
+	$node->adjust_conf('postgresql.conf', 'test_custom_var_stats.kind',
+		"'$kind_name'");
+	$node->restart;
+
+	# Create entries for variable-sized stats.
+	$node->safe_psql('postgres',
+		q(select test_custom_stats_var_create('entry1', 'Test entry 1')));
+	$node->safe_psql('postgres',
+		q(select test_custom_stats_var_create('entry2', 'Test entry 2')));
+	$node->safe_psql('postgres',
+		q(select test_custom_stats_var_create('entry3', 'Test entry 3')));
+	$node->safe_psql('postgres',
+		q(select test_custom_stats_var_create('entry4', 'Test entry 4')));
+
+	# Update counters: entry1=2, entry2=3, entry3=2, entry4=3
+	$node->safe_psql(
+		'postgres',
+		q(select test_custom_stats_var_update(name)
+		  from (values ('entry1', 2), ('entry2', 3),
+		               ('entry3', 2), ('entry4', 3)) as stats(name, n),
+		       generate_series(1, n)));
+
+	# Test data reports.
+	$result = $node->safe_psql('postgres',
+		q(select * from test_custom_stats_var_report('entry1')));
+	is( $result,
+		"entry1|2|Test entry 1",
+		"variable-sized stats of entry1 for $kind_name");
+
+	$result = $node->safe_psql('postgres',
+		q(select * from test_custom_stats_var_report('entry2')));
+	is( $result,
+		"entry2|3|Test entry 2",
+		"variable-sized stats of entry2 for $kind_name");
+
+	$result = $node->safe_psql('postgres',
+		q(select * from test_custom_stats_var_report('entry3')));
+	is( $result,
+		"entry3|2|Test entry 3",
+		"variable-sized stats of entry3 for $kind_name");
+
+	$result = $node->safe_psql('postgres',
+		q(select * from test_custom_stats_var_report('entry4')));
+	is( $result,
+		"entry4|3|Test entry 4",
+		"variable-sized stats of entry4 for $kind_name");
+
+	# Test drop of variable-sized stats.
+	$node->safe_psql('postgres',
+		q(select * from test_custom_stats_var_drop('entry3')));
+	$result = $node->safe_psql('postgres',
+		q(select * from test_custom_stats_var_report('entry3')));
+	is($result, "", "entry3 not found after drop for $kind_name");
+	$node->safe_psql('postgres',
+		q(select * from test_custom_stats_var_drop('entry4')));
+	$result = $node->safe_psql('postgres',
+		q(select * from test_custom_stats_var_report('entry4')));
+	is($result, "", "entry4 not found after drop for $kind_name");
+
+	# Test persistence across clean restart.
+	$node->stop();
+	$node->start();
+
+	$result = $node->safe_psql('postgres',
+		q(select * from test_custom_stats_var_report('entry1')));
+	is( $result,
+		"entry1|2|Test entry 1",
+		"variable-sized stats of entry1 persist after clean restart for $kind_name"
+	);
+
+	$result = $node->safe_psql('postgres',
+		q(select * from test_custom_stats_var_report('entry2')));
+	is( $result,
+		"entry2|3|Test entry 2",
+		"variable-sized stats of entry2 persist after clean restart for $kind_name"
+	);
+
+	# Test reset of variable-sized stats.
+	$node->safe_psql('postgres', q(select test_custom_stats_var_reset()));
+	$result = $node->safe_psql('postgres',
+		q(select * from test_custom_stats_var_report('entry1')));
+	is( $result,
+		"entry1|0|Test entry 1",
+		"variable-sized stats of entry1 reset for $kind_name");
+
+	$result = $node->safe_psql('postgres',
+		q(select * from test_custom_stats_var_report('entry2')));
+	is( $result,
+		"entry2|0|Test entry 2",
+		"variable-sized stats of entry2 reset for $kind_name");
+
+	# Test loss after crash recovery.
+	$node->stop('immediate');
+	$node->start;
+
+	$result = $node->safe_psql('postgres',
+		q(select * from test_custom_stats_var_report('entry1')));
+	is($result, "",
+		"variable-sized stats of entry1 lost after crash recovery for $kind_name"
+	);
+	$result = $node->safe_psql('postgres',
+		q(select * from test_custom_stats_var_report('entry2')));
+	is($result, "",
+		"variable-sized stats of entry2 lost after crash recovery for $kind_name"
+	);
+}
+
+# fixed-sized stats are updated 3 times, so the report below should match.
 $node->safe_psql('postgres', q(select test_custom_stats_fixed_update()));
 $node->safe_psql('postgres', q(select test_custom_stats_fixed_update()));
 $node->safe_psql('postgres', q(select test_custom_stats_fixed_update()));
 
-# Test data reports.
 $result = $node->safe_psql('postgres',
-	q(select * from test_custom_stats_var_report('entry1')));
-is( $result,
-	"entry1|2|Test entry 1",
-	"report for variable-sized data of entry1");
+	q(select numcalls from test_custom_stats_fixed_report()));
+is($result, "3", "report for fixed-sized stats");
 
-$result = $node->safe_psql('postgres',
-	q(select * from test_custom_stats_var_report('entry2')));
-is( $result,
-	"entry2|3|Test entry 2",
-	"report for variable-sized data of entry2");
-
-$result = $node->safe_psql('postgres',
-	q(select * from test_custom_stats_var_report('entry3')));
-is( $result,
-	"entry3|2|Test entry 3",
-	"report for variable-sized data of entry3");
-
-$result = $node->safe_psql('postgres',
-	q(select * from test_custom_stats_var_report('entry4')));
-is( $result,
-	"entry4|3|Test entry 4",
-	"report for variable-sized data of entry4");
-
-$result = $node->safe_psql('postgres',
-	q(select * from test_custom_stats_fixed_report()));
-is($result, "3|", "report for fixed-sized stats");
-
-# Test drop of variable-sized stats.
-$node->safe_psql('postgres',
-	q(select * from test_custom_stats_var_drop('entry3')));
-$result = $node->safe_psql('postgres',
-	q(select * from test_custom_stats_var_report('entry3')));
-is($result, "", "entry3 not found after drop");
-$node->safe_psql('postgres',
-	q(select * from test_custom_stats_var_drop('entry4')));
-$result = $node->safe_psql('postgres',
-	q(select * from test_custom_stats_var_report('entry4')));
-is($result, "", "entry4 not found after drop");
-
-# Test persistence across clean restart.
+# Test persistence of fixed-sized stats across clean restart.
 $node->stop();
 $node->start();
 
 $result = $node->safe_psql('postgres',
-	q(select * from test_custom_stats_var_report('entry1')));
-is( $result,
-	"entry1|2|Test entry 1",
-	"variable-sized stats persist after clean restart");
+	q(select numcalls from test_custom_stats_fixed_report()));
+is($result, "3", "fixed-sized stats persist after clean restart");
 
-$result = $node->safe_psql('postgres',
-	q(select * from test_custom_stats_var_report('entry2')));
-is( $result,
-	"entry2|3|Test entry 2",
-	"variable-sized stats persist after clean restart");
-
-$result = $node->safe_psql('postgres',
-	q(select * from test_custom_stats_fixed_report()));
-is($result, "3|", "fixed-sized stats persist after clean restart");
-
-# Test persistence after crash recovery.
+# Test fixed-sized stats after crash recovery.
 $node->stop('immediate');
 $node->start;
 
-$result = $node->safe_psql('postgres',
-	q(select * from test_custom_stats_var_report('entry1')));
-is($result, "", "variable-sized stats of entry1 lost after crash recovery");
-$result = $node->safe_psql('postgres',
-	q(select * from test_custom_stats_var_report('entry2')));
-is($result, "", "variable-sized stats of entry2 lost after crash recovery");
-
 # Crash recovery sets the reset timestamp.
 $result = $node->safe_psql('postgres',
 	q(select numcalls from test_custom_stats_fixed_report() where stats_reset is not null)
diff --git a/src/test/modules/test_custom_stats/test_custom_var_stats--1.0.sql b/src/test/modules/test_custom_stats/test_custom_var_stats--1.0.sql
index 5ed8cfc2dcf..2b92316679f 100644
--- a/src/test/modules/test_custom_stats/test_custom_var_stats--1.0.sql
+++ b/src/test/modules/test_custom_stats/test_custom_var_stats--1.0.sql
@@ -18,6 +18,11 @@ RETURNS void
 AS 'MODULE_PATHNAME', 'test_custom_stats_var_drop'
 LANGUAGE C STRICT PARALLEL UNSAFE;
 
+CREATE FUNCTION test_custom_stats_var_reset()
+RETURNS void
+AS 'MODULE_PATHNAME', 'test_custom_stats_var_reset'
+LANGUAGE C STRICT PARALLEL UNSAFE;
+
 CREATE FUNCTION test_custom_stats_var_report(INOUT name TEXT,
                                              OUT calls BIGINT,
                                              OUT description TEXT)
diff --git a/src/test/modules/test_custom_stats/test_custom_var_stats.c b/src/test/modules/test_custom_stats/test_custom_var_stats.c
index ec06d5e1fdb..ba6486fff6a 100644
--- a/src/test/modules/test_custom_stats/test_custom_var_stats.c
+++ b/src/test/modules/test_custom_stats/test_custom_var_stats.c
@@ -18,6 +18,7 @@
 #include "storage/dsm_registry.h"
 #include "storage/fd.h"
 #include "utils/builtins.h"
+#include "utils/guc.h"
 #include "utils/pgstat_internal.h"
 
 PG_MODULE_MAGIC_EXT(
@@ -111,7 +112,27 @@ static void test_custom_stats_var_finish(PgStat_StatsFileOp status);
  *--------------------------------------------------------------------------
  */
 
-static const PgStat_KindInfo custom_stats = {
+/* Which variable-sized custom stats kind definition to register. */
+static char *test_custom_stats_kind = "test_custom_var_stats";
+
+static const PgStat_KindInfo custom_stats_own_hash = {
+	.name = "test_custom_var_stats_own_hash",
+	.fixed_amount = false,		/* variable number of entries */
+	.write_to_file = true,		/* persist across restarts */
+	.track_entry_count = true,	/* count active entries */
+	.accessed_across_databases = true,	/* global statistics */
+	.own_hash = true,			/* use dedicated dshash */
+	.shared_size = sizeof(PgStatShared_CustomVarEntry),
+	.shared_data_off = offsetof(PgStatShared_CustomVarEntry, stats),
+	.shared_data_len = sizeof(((PgStatShared_CustomVarEntry *) 0)->stats),
+	.pending_size = sizeof(PgStat_StatCustomVarEntry),
+	.flush_pending_cb = test_custom_stats_var_flush_pending_cb,
+	.to_serialized_data = test_custom_stats_var_to_serialized_data,
+	.from_serialized_data = test_custom_stats_var_from_serialized_data,
+	.finish = test_custom_stats_var_finish,
+};
+
+static const PgStat_KindInfo custom_stats_shared_hash = {
 	.name = "test_custom_var_stats",
 	.fixed_amount = false,		/* variable number of entries */
 	.write_to_file = true,		/* persist across restarts */
@@ -135,8 +156,27 @@ static const PgStat_KindInfo custom_stats = {
 void
 _PG_init(void)
 {
-	/* Register custom statistics kind */
-	pgstat_register_kind(PGSTAT_KIND_TEST_CUSTOM_VAR_STATS, &custom_stats);
+	DefineCustomStringVariable("test_custom_var_stats.kind",
+							   "Sets the custom variable stats kind to register.",
+							   NULL,
+							   &test_custom_stats_kind,
+							   "test_custom_var_stats",
+							   PGC_POSTMASTER,
+							   0,
+							   NULL, NULL, NULL);
+
+	if (strcmp(test_custom_stats_kind, custom_stats_shared_hash.name) == 0)
+		pgstat_register_kind(PGSTAT_KIND_TEST_CUSTOM_VAR_STATS,
+							 &custom_stats_shared_hash);
+	else if (strcmp(test_custom_stats_kind, custom_stats_own_hash.name) == 0)
+		pgstat_register_kind(PGSTAT_KIND_TEST_CUSTOM_VAR_STATS,
+							 &custom_stats_own_hash);
+	else
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("invalid value for parameter \"%s\": \"%s\"",
+						"test_custom_var_stats.kind",
+						test_custom_stats_kind)));
 }
 
 /*--------------------------------------------------------------------------
@@ -615,6 +655,19 @@ test_custom_stats_var_drop(PG_FUNCTION_ARGS)
 	PG_RETURN_VOID();
 }
 
+/*
+ * test_custom_stats_var_reset
+ *		Reset all custom statistic entries
+ */
+PG_FUNCTION_INFO_V1(test_custom_stats_var_reset);
+Datum
+test_custom_stats_var_reset(PG_FUNCTION_ARGS)
+{
+	pgstat_reset_of_kind(PGSTAT_KIND_TEST_CUSTOM_VAR_STATS);
+
+	PG_RETURN_VOID();
+}
+
 /*
  * test_custom_stats_var_report
  *		Retrieve custom statistic values