v14-0009-PoC-lazy-vacuum-integration.patch

From 2431edf71e7e22248af46588f554c47cd169cec7 Mon Sep 17 00:00:00 2001
From: Masahiko Sawada <sawada.mshk@gmail.com>
Date: Fri, 4 Nov 2022 14:14:42 +0900
Subject: [PATCH v14 9/9] PoC: lazy vacuum integration.

The patch includes:

* Introducing a new module, TIDStore, to store TID in radix tree.
* Integrating TIDStore with Lazy (parallel) vacuum.
---
 src/backend/access/common/Makefile    |   1 +
 src/backend/access/common/meson.build |   1 +
 src/backend/access/common/tidstore.c  | 531 ++++++++++++++++++++++++++
 src/backend/access/heap/vacuumlazy.c  | 170 +++------
 src/backend/catalog/system_views.sql  |   2 +-
 src/backend/commands/vacuum.c         |  76 +---
 src/backend/commands/vacuumparallel.c |  64 ++--
 src/backend/storage/lmgr/lwlock.c     |   2 +
 src/backend/utils/misc/guc_tables.c   |   2 +-
 src/include/access/tidstore.h         |  49 +++
 src/include/commands/progress.h       |   4 +-
 src/include/commands/vacuum.h         |  25 +-
 src/include/storage/lwlock.h          |   1 +
 src/test/regress/expected/rules.out   |   4 +-
 14 files changed, 696 insertions(+), 236 deletions(-)
 create mode 100644 src/backend/access/common/tidstore.c
 create mode 100644 src/include/access/tidstore.h

diff --git a/src/backend/access/common/Makefile b/src/backend/access/common/Makefile
index b9aff0ccfd..67b8cc6108 100644
--- a/src/backend/access/common/Makefile
+++ b/src/backend/access/common/Makefile
@@ -27,6 +27,7 @@ OBJS = \
 	syncscan.o \
 	toast_compression.o \
 	toast_internals.o \
+	tidstore.o \
 	tupconvert.o \
 	tupdesc.o
 
diff --git a/src/backend/access/common/meson.build b/src/backend/access/common/meson.build
index 857beaa32d..76265974b1 100644
--- a/src/backend/access/common/meson.build
+++ b/src/backend/access/common/meson.build
@@ -13,6 +13,7 @@ backend_sources += files(
   'syncscan.c',
   'toast_compression.c',
   'toast_internals.c',
+  'tidstore.c',
   'tupconvert.c',
   'tupdesc.c',
 )
diff --git a/src/backend/access/common/tidstore.c b/src/backend/access/common/tidstore.c
new file mode 100644
index 0000000000..770c4ab5bf
--- /dev/null
+++ b/src/backend/access/common/tidstore.c
@@ -0,0 +1,531 @@
+/*-------------------------------------------------------------------------
+ *
+ * tidstore.c
+ *		TID (ItemPointer) storage implementation.
+ *
+ * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * IDENTIFICATION
+ *	  src/backend/access/common/tidstore.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/htup_details.h"
+#include "access/tidstore.h"
+#include "lib/radixtree.h"
+#include "port/pg_bitutils.h"
+#include "utils/dsa.h"
+#include "utils/memutils.h"
+#include "miscadmin.h"
+
+/* XXX only testing purpose during development, will be removed */
+#define XXX_DEBUG_TID_STORE 1
+
+/*
+ * For encoding purposes, item pointers are represented as a pair of 64-bit
+ * key and 64-bit value. We construct 64-bit unsigned integer that combines
+ * the block number and the offset number. The lowest 11 bits represent the
+ * offset number, and the next 32 bits are block number. That is, only 43
+ * bits are used:
+ *
+ * XXXXXXXX XXXYYYYY YYYYYYYY YYYYYYYY YYYYYYYY YYYuuuu
+ *
+ * X = bits used for offset number
+ * Y = bits used for block number
+ * u = unused bit
+ *
+ * 11 bits enough for the offset number, because MaxHeapTuplesPerPage < 2^11
+ * on all supported block sizes (TIDSTORE_OFFSET_NBITS). We are frugal with
+ * the bits, because smaller keys could help keeping the radix tree shallow.
+ *
+ * XXX: If we want to support other table AMs that want to use the full range
+ * of possible offset numbers, we'll need to change this.
+ *
+ * The 64-bit value is the bitmap representation of the lowest 6 bits, and
+ * the rest 37 bits are used as the key:
+ *
+ * value = bitmap representation of XXXXXX
+ * key = XXXXXYYY YYYYYYYY YYYYYYYY YYYYYYYY YYYYYuu
+ */
+#define TIDSTORE_OFFSET_NBITS 11
+#define TIDSTORE_VALUE_NBITS 6	/* log(sizeof(uint64) * BITS_PER_BYTE, 2) */
+
+/* Get block number from the key */
+#define KEY_GET_BLKNO(key) \
+	((BlockNumber) ((key) >> (TIDSTORE_OFFSET_NBITS - TIDSTORE_VALUE_NBITS)))
+
+struct TIDStore
+{
+	/* main storage for TID */
+	radix_tree	*tree;
+
+	/* # of tids in TIDStore */
+	int	num_tids;
+
+	/* maximum bytes TIDStore can consume */
+	uint64	max_bytes;
+
+	/* DSA area and handle for shared TIDStore */
+	rt_handle	handle;
+	dsa_area	*area;
+
+#if defined(USE_ASSERT_CHECKING) && defined(XXX_DEBUG_TID_STORE)
+	uint64	max_items;
+	ItemPointer	itemptrs;
+	uint64	nitems;
+#endif
+};
+
+/* Iterator for TDIStore */
+typedef struct TIDStoreIter
+{
+	TIDStore	*ts;
+
+	/* iterator of radix tree */
+	rt_iter		*tree_iter;
+
+	bool		finished;
+
+	/* save for the next iteration */
+	uint64		next_key;
+	uint64		next_val;
+
+	/* output for the caller */
+	TIDStoreIterResult result;
+
+#ifdef USE_ASSERT_CHECKING
+	uint64		itemptrs_index;
+	int	prev_index;
+#endif
+} TIDStoreIter;
+
+static void tidstore_iter_extract_tids(TIDStoreIter *iter, uint64 key, uint64 val);
+static inline uint64 tid_to_key_off(ItemPointer tid, uint32 *off);
+
+#if defined(USE_ASSERT_CHECKING) && defined(XXX_DEBUG_TID_STORE)
+/*
+ * Comparator routines for use with qsort() and bsearch().
+ */
+static int
+vac_cmp_itemptr(const void *left, const void *right)
+{
+	BlockNumber lblk,
+		rblk;
+	OffsetNumber loff,
+		roff;
+
+	lblk = ItemPointerGetBlockNumber((ItemPointer) left);
+	rblk = ItemPointerGetBlockNumber((ItemPointer) right);
+
+	if (lblk < rblk)
+		return -1;
+	if (lblk > rblk)
+		return 1;
+
+	loff = ItemPointerGetOffsetNumber((ItemPointer) left);
+	roff = ItemPointerGetOffsetNumber((ItemPointer) right);
+
+	if (loff < roff)
+		return -1;
+	if (loff > roff)
+		return 1;
+
+	return 0;
+}
+
+static void
+verify_iter_tids(TIDStoreIter *iter)
+{
+	uint64 index = iter->prev_index;
+	TIDStoreIterResult *result = &(iter->result);
+
+	if (iter->ts->itemptrs == NULL)
+		return;
+
+	Assert(index <= iter->ts->nitems);
+
+	for (int i = 0; i < result->num_offsets; i++)
+	{
+		ItemPointerData tid;
+
+		ItemPointerSetBlockNumber(&tid, result->blkno);
+		ItemPointerSetOffsetNumber(&tid, result->offsets[i]);
+
+		Assert(ItemPointerEquals(&iter->ts->itemptrs[index++], &tid));
+	}
+
+	iter->prev_index = iter->itemptrs_index;
+}
+
+static void
+dump_itemptrs(TIDStore *ts)
+{
+	StringInfoData buf;
+
+	if (ts->itemptrs == NULL)
+		return;
+
+	initStringInfo(&buf);
+	for (int i = 0; i < ts->nitems; i++)
+	{
+		appendStringInfo(&buf, "(%d,%d) ",
+						 ItemPointerGetBlockNumber(&(ts->itemptrs[i])),
+						 ItemPointerGetOffsetNumber(&(ts->itemptrs[i])));
+	}
+	elog(WARNING, "--- dump (" UINT64_FORMAT " items) ---", ts->nitems);
+	elog(WARNING, "%s\n", buf.data);
+}
+
+#endif
+
+/*
+ * Create a TIDStore. The returned object is allocated in backend-local memory.
+ * The radix tree for storage is allocated in DSA area is 'area' is non-NULL.
+ */
+TIDStore *
+tidstore_create(uint64 max_bytes, dsa_area *area)
+{
+	TIDStore	*ts;
+
+	ts = palloc0(sizeof(TIDStore));
+
+	ts->tree = rt_create(CurrentMemoryContext, area);
+	ts->area = area;
+	ts->max_bytes = max_bytes;
+
+	if (area != NULL)
+		ts->handle = rt_get_handle(ts->tree);
+
+#if defined(USE_ASSERT_CHECKING) && defined(XXX_DEBUG_TID_STORE)
+#define MAXDEADITEMS(avail_mem) \
+	(avail_mem / sizeof(ItemPointerData))
+
+	if (area == NULL)
+	{
+		ts->max_items = MAXDEADITEMS(maintenance_work_mem * 1024);
+		ts->itemptrs = (ItemPointer) palloc0(sizeof(ItemPointerData) * ts->max_items);
+		ts->nitems = 0;
+	}
+#endif
+
+	return ts;
+}
+
+/* Attach to the shared TIDStore using a handle */
+TIDStore *
+tidstore_attach(dsa_area *area, rt_handle handle)
+{
+	TIDStore *ts;
+
+	Assert(area != NULL);
+	Assert(DsaPointerIsValid(handle));
+
+	ts = palloc0(sizeof(TIDStore));
+	ts->tree = rt_attach(area, handle);
+
+	return ts;
+}
+
+/*
+ * Detach from a TIDStore. This detaches from radix tree and frees the
+ * backend-local resources.
+ */
+void
+tidstore_detach(TIDStore *ts)
+{
+	rt_detach(ts->tree);
+	pfree(ts);
+}
+
+void
+tidstore_free(TIDStore *ts)
+{
+#if defined(USE_ASSERT_CHECKING) && defined(XXX_DEBUG_TID_STORE)
+	if (ts->itemptrs)
+		pfree(ts->itemptrs);
+#endif
+
+	rt_free(ts->tree);
+	pfree(ts);
+}
+
+void
+tidstore_reset(TIDStore *ts)
+{
+	dsa_area *area = ts->area;
+
+	/* Reset the statistics */
+	ts->num_tids = 0;
+
+	/* Free the radix tree */
+	rt_free(ts->tree);
+
+	if (ts->area)
+		dsa_trim(area);
+
+	ts->tree = rt_create(CurrentMemoryContext, area);
+#if defined(USE_ASSERT_CHECKING) && defined(XXX_DEBUG_TID_STORE)
+	ts->nitems = 0;
+#endif
+}
+
+/* Add TIDs to TIDStore */
+void
+tidstore_add_tids(TIDStore *ts, BlockNumber blkno, OffsetNumber *offsets,
+				  int num_offsets)
+{
+	uint64 last_key = PG_UINT64_MAX;
+	uint64 key;
+	uint64 val = 0;
+	ItemPointerData tid;
+
+	ItemPointerSetBlockNumber(&tid, blkno);
+
+	for (int i = 0; i < num_offsets; i++)
+	{
+		uint32	off;
+
+		ItemPointerSetOffsetNumber(&tid, offsets[i]);
+
+		key = tid_to_key_off(&tid, &off);
+
+		if (last_key != PG_UINT64_MAX && last_key != key)
+		{
+			rt_set(ts->tree, last_key, val);
+			val = 0;
+		}
+
+		last_key = key;
+		val |= UINT64CONST(1) << off;
+		ts->num_tids++;
+
+#if defined(USE_ASSERT_CHECKING) && defined(XXX_DEBUG_TID_STORE)
+		if (ts->itemptrs)
+		{
+			if (ts->nitems >= ts->max_items)
+			{
+				ts->max_items *= 2;
+				ts->itemptrs = repalloc(ts->itemptrs, sizeof(ItemPointerData) * ts->max_items);
+			}
+
+			Assert(ts->nitems < ts->max_items);
+			ItemPointerSetBlockNumber(&(ts->itemptrs[ts->nitems]), blkno);
+			ItemPointerSetOffsetNumber(&(ts->itemptrs[ts->nitems]), offsets[i]);
+			ts->nitems++;
+		}
+#endif
+	}
+
+	if (last_key != PG_UINT64_MAX)
+	{
+		rt_set(ts->tree, last_key, val);
+		val = 0;
+	}
+
+#if defined(USE_ASSERT_CHECKING) && defined(XXX_DEBUG_TID_STORE)
+	if (ts->itemptrs)
+		Assert(ts->nitems == ts->num_tids);
+#endif
+}
+
+/* Return true if the given TID is present in TIDStore */
+bool
+tidstore_lookup_tid(TIDStore *ts, ItemPointer tid)
+{
+	uint64 key;
+	uint64 val;
+	uint32 off;
+	bool found;
+#if defined(USE_ASSERT_CHECKING) && defined(XXX_DEBUG_TID_STORE)
+	bool found_assert;
+#endif
+
+	key = tid_to_key_off(tid, &off);
+
+	found = rt_search(ts->tree, key, &val);
+
+#if defined(USE_ASSERT_CHECKING) && defined(XXX_DEBUG_TID_STORE)
+	if (ts->itemptrs)
+		found_assert = bsearch((void *) tid,
+							   (void *) ts->itemptrs,
+							   ts->nitems,
+							   sizeof(ItemPointerData),
+							   vac_cmp_itemptr) != NULL;
+#endif
+
+	if (!found)
+	{
+#if defined(USE_ASSERT_CHECKING) && defined(XXX_DEBUG_TID_STORE)
+		if (ts->itemptrs)
+			Assert(!found_assert);
+#endif
+		return false;
+	}
+
+	found = (val & (UINT64CONST(1) << off)) != 0;
+
+#if defined(USE_ASSERT_CHECKING) && defined(XXX_DEBUG_TID_STORE)
+
+	if (ts->itemptrs && found != found_assert)
+	{
+		elog(WARNING, "tid (%d,%d)\n",
+				ItemPointerGetBlockNumber(tid),
+				ItemPointerGetOffsetNumber(tid));
+		dump_itemptrs(ts);
+	}
+
+	if (ts->itemptrs)
+		Assert(found == found_assert);
+
+#endif
+	return found;
+}
+
+TIDStoreIter *
+tidstore_begin_iterate(TIDStore *ts)
+{
+	TIDStoreIter *iter;
+
+	iter = palloc0(sizeof(TIDStoreIter));
+	iter->ts = ts;
+	iter->tree_iter = rt_begin_iterate(ts->tree);
+	iter->result.blkno = InvalidBlockNumber;
+
+#if defined(USE_ASSERT_CHECKING) && defined(XXX_DEBUG_TID_STORE)
+	iter->itemptrs_index = 0;
+#endif
+
+	return iter;
+}
+
+TIDStoreIterResult *
+tidstore_iterate_next(TIDStoreIter *iter)
+{
+	uint64 key;
+	uint64 val;
+	TIDStoreIterResult *result = &(iter->result);
+
+	if (iter->finished)
+		return NULL;
+
+	if (BlockNumberIsValid(result->blkno))
+	{
+		result->num_offsets = 0;
+		tidstore_iter_extract_tids(iter, iter->next_key, iter->next_val);
+	}
+
+	while (rt_iterate_next(iter->tree_iter, &key, &val))
+	{
+		BlockNumber blkno;
+
+		blkno = KEY_GET_BLKNO(key);
+
+		if (BlockNumberIsValid(result->blkno) && result->blkno != blkno)
+		{
+			/*
+			 * Remember the key-value pair for the next block for the
+			 * next iteration.
+			 */
+			iter->next_key = key;
+			iter->next_val = val;
+
+#if defined(USE_ASSERT_CHECKING) && defined(XXX_DEBUG_TID_STORE)
+			verify_iter_tids(iter);
+#endif
+			return result;
+		}
+
+		/* Collect tids extracted from the key-value pair */
+		tidstore_iter_extract_tids(iter, key, val);
+	}
+
+#if defined(USE_ASSERT_CHECKING) && defined(XXX_DEBUG_TID_STORE)
+	verify_iter_tids(iter);
+#endif
+
+	iter->finished = true;
+	return result;
+}
+
+uint64
+tidstore_num_tids(TIDStore *ts)
+{
+	return ts->num_tids;
+}
+
+bool
+tidstore_is_full(TIDStore *ts)
+{
+	return ((sizeof(TIDStore) + rt_memory_usage(ts->tree)) > ts->max_bytes);
+}
+
+uint64
+tidstore_max_memory(TIDStore *ts)
+{
+	return ts->max_bytes;
+}
+
+uint64
+tidstore_memory_usage(TIDStore *ts)
+{
+	return (uint64) sizeof(TIDStore) + rt_memory_usage(ts->tree);
+}
+
+/*
+ * Get a handle that can be used by other processes to attach to this TIDStore
+ */
+tidstore_handle
+tidstore_get_handle(TIDStore *ts)
+{
+	return rt_get_handle(ts->tree);
+}
+
+/* Extract TIDs from key-value pair */
+static void
+tidstore_iter_extract_tids(TIDStoreIter *iter, uint64 key, uint64 val)
+{
+	TIDStoreIterResult *result = (&iter->result);
+
+	for (int i = 0; i < sizeof(uint64) * BITS_PER_BYTE; i++)
+	{
+		uint64	tid_i;
+		OffsetNumber	off;
+
+		if ((val & (UINT64CONST(1) << i)) == 0)
+			continue;
+
+		tid_i = key << TIDSTORE_VALUE_NBITS;
+		tid_i |= i;
+
+		off = tid_i & ((UINT64CONST(1) << TIDSTORE_OFFSET_NBITS) - 1);
+		result->offsets[result->num_offsets++] = off;
+
+#if defined(USE_ASSERT_CHECKING) && defined(XXX_DEBUG_TID_STORE)
+		iter->itemptrs_index++;
+#endif
+	}
+
+	result->blkno = KEY_GET_BLKNO(key);
+}
+
+/*
+ * Encode a TID to key and val.
+ */
+static inline uint64
+tid_to_key_off(ItemPointer tid, uint32 *off)
+{
+	uint64 upper;
+	uint64 tid_i;
+
+	tid_i = ItemPointerGetOffsetNumber(tid);
+	tid_i |= (uint64) ItemPointerGetBlockNumber(tid) << TIDSTORE_OFFSET_NBITS;
+
+	*off = tid_i & ((1 << TIDSTORE_VALUE_NBITS) - 1);
+	upper = tid_i >> TIDSTORE_VALUE_NBITS;
+	Assert(*off < (sizeof(uint64) * BITS_PER_BYTE));
+
+	return upper;
+}
diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c
index d59711b7ec..24c1dc7099 100644
--- a/src/backend/access/heap/vacuumlazy.c
+++ b/src/backend/access/heap/vacuumlazy.c
@@ -40,6 +40,7 @@
 #include "access/heapam_xlog.h"
 #include "access/htup_details.h"
 #include "access/multixact.h"
+#include "access/tidstore.h"
 #include "access/transam.h"
 #include "access/visibilitymap.h"
 #include "access/xact.h"
@@ -194,7 +195,7 @@ typedef struct LVRelState
 	 * lazy_vacuum_heap_rel, which marks the same LP_DEAD line pointers as
 	 * LP_UNUSED during second heap pass.
 	 */
-	VacDeadItems *dead_items;	/* TIDs whose index tuples we'll delete */
+	TIDStore *dead_items;	/* TIDs whose index tuples we'll delete */
 	BlockNumber rel_pages;		/* total number of pages */
 	BlockNumber scanned_pages;	/* # pages examined (not skipped via VM) */
 	BlockNumber removed_pages;	/* # pages removed by relation truncation */
@@ -265,8 +266,9 @@ static bool lazy_scan_noprune(LVRelState *vacrel, Buffer buf,
 static void lazy_vacuum(LVRelState *vacrel);
 static bool lazy_vacuum_all_indexes(LVRelState *vacrel);
 static void lazy_vacuum_heap_rel(LVRelState *vacrel);
-static int	lazy_vacuum_heap_page(LVRelState *vacrel, BlockNumber blkno,
-								  Buffer buffer, int index, Buffer *vmbuffer);
+static void	lazy_vacuum_heap_page(LVRelState *vacrel, BlockNumber blkno,
+								  OffsetNumber *offsets, int num_offsets,
+								  Buffer buffer, Buffer *vmbuffer);
 static bool lazy_check_wraparound_failsafe(LVRelState *vacrel);
 static void lazy_cleanup_all_indexes(LVRelState *vacrel);
 static IndexBulkDeleteResult *lazy_vacuum_one_index(Relation indrel,
@@ -853,21 +855,21 @@ lazy_scan_heap(LVRelState *vacrel)
 				next_unskippable_block,
 				next_failsafe_block = 0,
 				next_fsm_block_to_vacuum = 0;
-	VacDeadItems *dead_items = vacrel->dead_items;
+	TIDStore *dead_items = vacrel->dead_items;
 	Buffer		vmbuffer = InvalidBuffer;
 	bool		next_unskippable_allvis,
 				skipping_current_range;
 	const int	initprog_index[] = {
 		PROGRESS_VACUUM_PHASE,
 		PROGRESS_VACUUM_TOTAL_HEAP_BLKS,
-		PROGRESS_VACUUM_MAX_DEAD_TUPLES
+		PROGRESS_VACUUM_MAX_DEAD_TUPLE_BYTES
 	};
 	int64		initprog_val[3];
 
 	/* Report that we're scanning the heap, advertising total # of blocks */
 	initprog_val[0] = PROGRESS_VACUUM_PHASE_SCAN_HEAP;
 	initprog_val[1] = rel_pages;
-	initprog_val[2] = dead_items->max_items;
+	initprog_val[2] = tidstore_max_memory(vacrel->dead_items);
 	pgstat_progress_update_multi_param(3, initprog_index, initprog_val);
 
 	/* Set up an initial range of skippable blocks using the visibility map */
@@ -937,8 +939,8 @@ lazy_scan_heap(LVRelState *vacrel)
 		 * dead_items TIDs, pause and do a cycle of vacuuming before we tackle
 		 * this page.
 		 */
-		Assert(dead_items->max_items >= MaxHeapTuplesPerPage);
-		if (dead_items->max_items - dead_items->num_items < MaxHeapTuplesPerPage)
+		/* XXX: should not allow tidstore to grow beyond max_bytes */
+		if (tidstore_is_full(vacrel->dead_items))
 		{
 			/*
 			 * Before beginning index vacuuming, we release any pin we may
@@ -1070,11 +1072,18 @@ lazy_scan_heap(LVRelState *vacrel)
 			if (prunestate.has_lpdead_items)
 			{
 				Size		freespace;
+				TIDStoreIter *iter;
+				TIDStoreIterResult *result;
 
-				lazy_vacuum_heap_page(vacrel, blkno, buf, 0, &vmbuffer);
+				iter = tidstore_begin_iterate(vacrel->dead_items);
+				result = tidstore_iterate_next(iter);
+				lazy_vacuum_heap_page(vacrel, blkno, result->offsets, result->num_offsets,
+									  buf, &vmbuffer);
+				Assert(!tidstore_iterate_next(iter));
+				pfree(iter);
 
 				/* Forget the LP_DEAD items that we just vacuumed */
-				dead_items->num_items = 0;
+				tidstore_reset(dead_items);
 
 				/*
 				 * Periodically perform FSM vacuuming to make newly-freed
@@ -1111,7 +1120,7 @@ lazy_scan_heap(LVRelState *vacrel)
 			 * with prunestate-driven visibility map and FSM steps (just like
 			 * the two-pass strategy).
 			 */
-			Assert(dead_items->num_items == 0);
+			Assert(tidstore_num_tids(dead_items) == 0);
 		}
 
 		/*
@@ -1264,7 +1273,7 @@ lazy_scan_heap(LVRelState *vacrel)
 	 * Do index vacuuming (call each index's ambulkdelete routine), then do
 	 * related heap vacuuming
 	 */
-	if (dead_items->num_items > 0)
+	if (tidstore_num_tids(dead_items) > 0)
 		lazy_vacuum(vacrel);
 
 	/*
@@ -1863,25 +1872,16 @@ retry:
 	 */
 	if (lpdead_items > 0)
 	{
-		VacDeadItems *dead_items = vacrel->dead_items;
-		ItemPointerData tmp;
+		TIDStore *dead_items = vacrel->dead_items;
 
 		Assert(!prunestate->all_visible);
 		Assert(prunestate->has_lpdead_items);
 
 		vacrel->lpdead_item_pages++;
+		tidstore_add_tids(dead_items, blkno, deadoffsets, lpdead_items);
 
-		ItemPointerSetBlockNumber(&tmp, blkno);
-
-		for (int i = 0; i < lpdead_items; i++)
-		{
-			ItemPointerSetOffsetNumber(&tmp, deadoffsets[i]);
-			dead_items->items[dead_items->num_items++] = tmp;
-		}
-
-		Assert(dead_items->num_items <= dead_items->max_items);
-		pgstat_progress_update_param(PROGRESS_VACUUM_NUM_DEAD_TUPLES,
-									 dead_items->num_items);
+		pgstat_progress_update_param(PROGRESS_VACUUM_DEAD_TUPLE_BYTES,
+									 tidstore_memory_usage(dead_items));
 	}
 
 	/* Finally, add page-local counts to whole-VACUUM counts */
@@ -2088,8 +2088,7 @@ lazy_scan_noprune(LVRelState *vacrel,
 	}
 	else
 	{
-		VacDeadItems *dead_items = vacrel->dead_items;
-		ItemPointerData tmp;
+		TIDStore *dead_items = vacrel->dead_items;
 
 		/*
 		 * Page has LP_DEAD items, and so any references/TIDs that remain in
@@ -2098,17 +2097,10 @@ lazy_scan_noprune(LVRelState *vacrel,
 		 */
 		vacrel->lpdead_item_pages++;
 
-		ItemPointerSetBlockNumber(&tmp, blkno);
+		tidstore_add_tids(dead_items, blkno, deadoffsets, lpdead_items);
 
-		for (int i = 0; i < lpdead_items; i++)
-		{
-			ItemPointerSetOffsetNumber(&tmp, deadoffsets[i]);
-			dead_items->items[dead_items->num_items++] = tmp;
-		}
-
-		Assert(dead_items->num_items <= dead_items->max_items);
-		pgstat_progress_update_param(PROGRESS_VACUUM_NUM_DEAD_TUPLES,
-									 dead_items->num_items);
+		pgstat_progress_update_param(PROGRESS_VACUUM_DEAD_TUPLE_BYTES,
+									 tidstore_memory_usage(dead_items));
 
 		vacrel->lpdead_items += lpdead_items;
 
@@ -2157,7 +2149,7 @@ lazy_vacuum(LVRelState *vacrel)
 	if (!vacrel->do_index_vacuuming)
 	{
 		Assert(!vacrel->do_index_cleanup);
-		vacrel->dead_items->num_items = 0;
+		tidstore_reset(vacrel->dead_items);
 		return;
 	}
 
@@ -2186,7 +2178,7 @@ lazy_vacuum(LVRelState *vacrel)
 		BlockNumber threshold;
 
 		Assert(vacrel->num_index_scans == 0);
-		Assert(vacrel->lpdead_items == vacrel->dead_items->num_items);
+		Assert(vacrel->lpdead_items == tidstore_num_tids(vacrel->dead_items));
 		Assert(vacrel->do_index_vacuuming);
 		Assert(vacrel->do_index_cleanup);
 
@@ -2213,8 +2205,8 @@ lazy_vacuum(LVRelState *vacrel)
 		 * cases then this may need to be reconsidered.
 		 */
 		threshold = (double) vacrel->rel_pages * BYPASS_THRESHOLD_PAGES;
-		bypass = (vacrel->lpdead_item_pages < threshold &&
-				  vacrel->lpdead_items < MAXDEADITEMS(32L * 1024L * 1024L));
+		bypass = (vacrel->lpdead_item_pages < threshold) &&
+			tidstore_memory_usage(vacrel->dead_items) < (32L * 1024L * 1024L);
 	}
 
 	if (bypass)
@@ -2259,7 +2251,7 @@ lazy_vacuum(LVRelState *vacrel)
 	 * Forget the LP_DEAD items that we just vacuumed (or just decided to not
 	 * vacuum)
 	 */
-	vacrel->dead_items->num_items = 0;
+	tidstore_reset(vacrel->dead_items);
 }
 
 /*
@@ -2331,7 +2323,7 @@ lazy_vacuum_all_indexes(LVRelState *vacrel)
 	 * place).
 	 */
 	Assert(vacrel->num_index_scans > 0 ||
-		   vacrel->dead_items->num_items == vacrel->lpdead_items);
+		   tidstore_num_tids(vacrel->dead_items) == vacrel->lpdead_items);
 	Assert(allindexes || vacrel->failsafe_active);
 
 	/*
@@ -2368,10 +2360,11 @@ lazy_vacuum_all_indexes(LVRelState *vacrel)
 static void
 lazy_vacuum_heap_rel(LVRelState *vacrel)
 {
-	int			index;
 	BlockNumber vacuumed_pages;
 	Buffer		vmbuffer = InvalidBuffer;
 	LVSavedErrInfo saved_err_info;
+	TIDStoreIter *iter;
+	TIDStoreIterResult *result;
 
 	Assert(vacrel->do_index_vacuuming);
 	Assert(vacrel->do_index_cleanup);
@@ -2388,8 +2381,8 @@ lazy_vacuum_heap_rel(LVRelState *vacrel)
 
 	vacuumed_pages = 0;
 
-	index = 0;
-	while (index < vacrel->dead_items->num_items)
+	iter = tidstore_begin_iterate(vacrel->dead_items);
+	while ((result = tidstore_iterate_next(iter)) != NULL)
 	{
 		BlockNumber tblk;
 		Buffer		buf;
@@ -2398,12 +2391,13 @@ lazy_vacuum_heap_rel(LVRelState *vacrel)
 
 		vacuum_delay_point();
 
-		tblk = ItemPointerGetBlockNumber(&vacrel->dead_items->items[index]);
+		tblk = result->blkno;
 		vacrel->blkno = tblk;
 		buf = ReadBufferExtended(vacrel->rel, MAIN_FORKNUM, tblk, RBM_NORMAL,
 								 vacrel->bstrategy);
 		LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
-		index = lazy_vacuum_heap_page(vacrel, tblk, buf, index, &vmbuffer);
+		lazy_vacuum_heap_page(vacrel, tblk, result->offsets, result->num_offsets,
+							  buf, &vmbuffer);
 
 		/* Now that we've vacuumed the page, record its available space */
 		page = BufferGetPage(buf);
@@ -2427,14 +2421,13 @@ lazy_vacuum_heap_rel(LVRelState *vacrel)
 	 * We set all LP_DEAD items from the first heap pass to LP_UNUSED during
 	 * the second heap pass.  No more, no less.
 	 */
-	Assert(index > 0);
 	Assert(vacrel->num_index_scans > 1 ||
-		   (index == vacrel->lpdead_items &&
+		   (tidstore_num_tids(vacrel->dead_items) == vacrel->lpdead_items &&
 			vacuumed_pages == vacrel->lpdead_item_pages));
 
 	ereport(DEBUG2,
-			(errmsg("table \"%s\": removed %lld dead item identifiers in %u pages",
-					vacrel->relname, (long long) index, vacuumed_pages)));
+			(errmsg("table \"%s\": removed " UINT64_FORMAT "dead item identifiers in %u pages",
+					vacrel->relname, tidstore_num_tids(vacrel->dead_items), vacuumed_pages)));
 
 	/* Revert to the previous phase information for error traceback */
 	restore_vacuum_error_info(vacrel, &saved_err_info);
@@ -2451,11 +2444,10 @@ lazy_vacuum_heap_rel(LVRelState *vacrel)
  * LP_DEAD item on the page.  The return value is the first index immediately
  * after all LP_DEAD items for the same page in the array.
  */
-static int
-lazy_vacuum_heap_page(LVRelState *vacrel, BlockNumber blkno, Buffer buffer,
-					  int index, Buffer *vmbuffer)
+static void
+lazy_vacuum_heap_page(LVRelState *vacrel, BlockNumber blkno, OffsetNumber *offsets,
+					  int num_offsets, Buffer buffer, Buffer *vmbuffer)
 {
-	VacDeadItems *dead_items = vacrel->dead_items;
 	Page		page = BufferGetPage(buffer);
 	OffsetNumber unused[MaxHeapTuplesPerPage];
 	int			uncnt = 0;
@@ -2474,16 +2466,11 @@ lazy_vacuum_heap_page(LVRelState *vacrel, BlockNumber blkno, Buffer buffer,
 
 	START_CRIT_SECTION();
 
-	for (; index < dead_items->num_items; index++)
+	for (int i = 0; i < num_offsets; i++)
 	{
-		BlockNumber tblk;
-		OffsetNumber toff;
 		ItemId		itemid;
+		OffsetNumber	toff = offsets[i];
 
-		tblk = ItemPointerGetBlockNumber(&dead_items->items[index]);
-		if (tblk != blkno)
-			break;				/* past end of tuples for this block */
-		toff = ItemPointerGetOffsetNumber(&dead_items->items[index]);
 		itemid = PageGetItemId(page, toff);
 
 		Assert(ItemIdIsDead(itemid) && !ItemIdHasStorage(itemid));
@@ -2563,7 +2550,6 @@ lazy_vacuum_heap_page(LVRelState *vacrel, BlockNumber blkno, Buffer buffer,
 
 	/* Revert to the previous phase information for error traceback */
 	restore_vacuum_error_info(vacrel, &saved_err_info);
-	return index;
 }
 
 /*
@@ -3065,46 +3051,6 @@ count_nondeletable_pages(LVRelState *vacrel, bool *lock_waiter_detected)
 	return vacrel->nonempty_pages;
 }
 
-/*
- * Returns the number of dead TIDs that VACUUM should allocate space to
- * store, given a heap rel of size vacrel->rel_pages, and given current
- * maintenance_work_mem setting (or current autovacuum_work_mem setting,
- * when applicable).
- *
- * See the comments at the head of this file for rationale.
- */
-static int
-dead_items_max_items(LVRelState *vacrel)
-{
-	int64		max_items;
-	int			vac_work_mem = IsAutoVacuumWorkerProcess() &&
-	autovacuum_work_mem != -1 ?
-	autovacuum_work_mem : maintenance_work_mem;
-
-	if (vacrel->nindexes > 0)
-	{
-		BlockNumber rel_pages = vacrel->rel_pages;
-
-		max_items = MAXDEADITEMS(vac_work_mem * 1024L);
-		max_items = Min(max_items, INT_MAX);
-		max_items = Min(max_items, MAXDEADITEMS(MaxAllocSize));
-
-		/* curious coding here to ensure the multiplication can't overflow */
-		if ((BlockNumber) (max_items / MaxHeapTuplesPerPage) > rel_pages)
-			max_items = rel_pages * MaxHeapTuplesPerPage;
-
-		/* stay sane if small maintenance_work_mem */
-		max_items = Max(max_items, MaxHeapTuplesPerPage);
-	}
-	else
-	{
-		/* One-pass case only stores a single heap page's TIDs at a time */
-		max_items = MaxHeapTuplesPerPage;
-	}
-
-	return (int) max_items;
-}
-
 /*
  * Allocate dead_items (either using palloc, or in dynamic shared memory).
  * Sets dead_items in vacrel for caller.
@@ -3115,11 +3061,9 @@ dead_items_max_items(LVRelState *vacrel)
 static void
 dead_items_alloc(LVRelState *vacrel, int nworkers)
 {
-	VacDeadItems *dead_items;
-	int			max_items;
-
-	max_items = dead_items_max_items(vacrel);
-	Assert(max_items >= MaxHeapTuplesPerPage);
+	int			vac_work_mem = IsAutoVacuumWorkerProcess() &&
+		autovacuum_work_mem != -1 ?
+		autovacuum_work_mem * 1024L : maintenance_work_mem * 1024L;
 
 	/*
 	 * Initialize state for a parallel vacuum.  As of now, only one worker can
@@ -3146,7 +3090,7 @@ dead_items_alloc(LVRelState *vacrel, int nworkers)
 		else
 			vacrel->pvs = parallel_vacuum_init(vacrel->rel, vacrel->indrels,
 											   vacrel->nindexes, nworkers,
-											   max_items,
+											   vac_work_mem,
 											   vacrel->verbose ? INFO : DEBUG2,
 											   vacrel->bstrategy);
 
@@ -3159,11 +3103,7 @@ dead_items_alloc(LVRelState *vacrel, int nworkers)
 	}
 
 	/* Serial VACUUM case */
-	dead_items = (VacDeadItems *) palloc(vac_max_items_to_alloc_size(max_items));
-	dead_items->max_items = max_items;
-	dead_items->num_items = 0;
-
-	vacrel->dead_items = dead_items;
+	vacrel->dead_items = tidstore_create(vac_work_mem, NULL);
 }
 
 /*
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 2d8104b090..bc42144f08 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1165,7 +1165,7 @@ CREATE VIEW pg_stat_progress_vacuum AS
                       END AS phase,
         S.param2 AS heap_blks_total, S.param3 AS heap_blks_scanned,
         S.param4 AS heap_blks_vacuumed, S.param5 AS index_vacuum_count,
-        S.param6 AS max_dead_tuples, S.param7 AS num_dead_tuples
+        S.param6 AS max_dead_tuple_bytes, S.param7 AS dead_tuple_bytes
     FROM pg_stat_get_progress_info('VACUUM') AS S
         LEFT JOIN pg_database D ON S.datid = D.oid;
 
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index 293b84bbca..7f5776fbf8 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -95,7 +95,6 @@ static bool vacuum_rel(Oid relid, RangeVar *relation, VacuumParams *params);
 static double compute_parallel_delay(void);
 static VacOptValue get_vacoptval_from_boolean(DefElem *def);
 static bool vac_tid_reaped(ItemPointer itemptr, void *state);
-static int	vac_cmp_itemptr(const void *left, const void *right);
 
 /*
  * Primary entry point for manual VACUUM and ANALYZE commands
@@ -2276,16 +2275,16 @@ get_vacoptval_from_boolean(DefElem *def)
  */
 IndexBulkDeleteResult *
 vac_bulkdel_one_index(IndexVacuumInfo *ivinfo, IndexBulkDeleteResult *istat,
-					  VacDeadItems *dead_items)
+					  TIDStore *dead_items)
 {
 	/* Do bulk deletion */
 	istat = index_bulk_delete(ivinfo, istat, vac_tid_reaped,
 							  (void *) dead_items);
 
 	ereport(ivinfo->message_level,
-			(errmsg("scanned index \"%s\" to remove %d row versions",
+			(errmsg("scanned index \"%s\" to remove " UINT64_FORMAT " row versions",
 					RelationGetRelationName(ivinfo->index),
-					dead_items->num_items)));
+					tidstore_num_tids(dead_items))));
 
 	return istat;
 }
@@ -2316,18 +2315,6 @@ vac_cleanup_one_index(IndexVacuumInfo *ivinfo, IndexBulkDeleteResult *istat)
 	return istat;
 }
 
-/*
- * Returns the total required space for VACUUM's dead_items array given a
- * max_items value.
- */
-Size
-vac_max_items_to_alloc_size(int max_items)
-{
-	Assert(max_items <= MAXDEADITEMS(MaxAllocSize));
-
-	return offsetof(VacDeadItems, items) + sizeof(ItemPointerData) * max_items;
-}
-
 /*
  *	vac_tid_reaped() -- is a particular tid deletable?
  *
@@ -2338,60 +2325,7 @@ vac_max_items_to_alloc_size(int max_items)
 static bool
 vac_tid_reaped(ItemPointer itemptr, void *state)
 {
-	VacDeadItems *dead_items = (VacDeadItems *) state;
-	int64		litem,
-				ritem,
-				item;
-	ItemPointer res;
-
-	litem = itemptr_encode(&dead_items->items[0]);
-	ritem = itemptr_encode(&dead_items->items[dead_items->num_items - 1]);
-	item = itemptr_encode(itemptr);
-
-	/*
-	 * Doing a simple bound check before bsearch() is useful to avoid the
-	 * extra cost of bsearch(), especially if dead items on the heap are
-	 * concentrated in a certain range.  Since this function is called for
-	 * every index tuple, it pays to be really fast.
-	 */
-	if (item < litem || item > ritem)
-		return false;
-
-	res = (ItemPointer) bsearch((void *) itemptr,
-								(void *) dead_items->items,
-								dead_items->num_items,
-								sizeof(ItemPointerData),
-								vac_cmp_itemptr);
-
-	return (res != NULL);
-}
-
-/*
- * Comparator routines for use with qsort() and bsearch().
- */
-static int
-vac_cmp_itemptr(const void *left, const void *right)
-{
-	BlockNumber lblk,
-				rblk;
-	OffsetNumber loff,
-				roff;
-
-	lblk = ItemPointerGetBlockNumber((ItemPointer) left);
-	rblk = ItemPointerGetBlockNumber((ItemPointer) right);
-
-	if (lblk < rblk)
-		return -1;
-	if (lblk > rblk)
-		return 1;
-
-	loff = ItemPointerGetOffsetNumber((ItemPointer) left);
-	roff = ItemPointerGetOffsetNumber((ItemPointer) right);
-
-	if (loff < roff)
-		return -1;
-	if (loff > roff)
-		return 1;
+	TIDStore *dead_items = (TIDStore *) state;
 
-	return 0;
+	return tidstore_lookup_tid(dead_items, itemptr);
 }
diff --git a/src/backend/commands/vacuumparallel.c b/src/backend/commands/vacuumparallel.c
index f26d796e52..429607d5fa 100644
--- a/src/backend/commands/vacuumparallel.c
+++ b/src/backend/commands/vacuumparallel.c
@@ -44,7 +44,7 @@
  * use small integers.
  */
 #define PARALLEL_VACUUM_KEY_SHARED			1
-#define PARALLEL_VACUUM_KEY_DEAD_ITEMS		2
+#define PARALLEL_VACUUM_KEY_DSA				2
 #define PARALLEL_VACUUM_KEY_QUERY_TEXT		3
 #define PARALLEL_VACUUM_KEY_BUFFER_USAGE	4
 #define PARALLEL_VACUUM_KEY_WAL_USAGE		5
@@ -103,6 +103,9 @@ typedef struct PVShared
 
 	/* Counter for vacuuming and cleanup */
 	pg_atomic_uint32 idx;
+
+	/* Handle of the shared TIDStore */
+	tidstore_handle	dead_items_handle;
 } PVShared;
 
 /* Status used during parallel index vacuum or cleanup */
@@ -166,7 +169,8 @@ struct ParallelVacuumState
 	PVIndStats *indstats;
 
 	/* Shared dead items space among parallel vacuum workers */
-	VacDeadItems *dead_items;
+	TIDStore *dead_items;
+	dsa_area *dead_items_area;
 
 	/* Points to buffer usage area in DSM */
 	BufferUsage *buffer_usage;
@@ -222,20 +226,23 @@ static void parallel_vacuum_error_callback(void *arg);
  */
 ParallelVacuumState *
 parallel_vacuum_init(Relation rel, Relation *indrels, int nindexes,
-					 int nrequested_workers, int max_items,
-					 int elevel, BufferAccessStrategy bstrategy)
+					 int nrequested_workers, int vac_work_mem,
+					 int elevel,
+					 BufferAccessStrategy bstrategy)
 {
 	ParallelVacuumState *pvs;
 	ParallelContext *pcxt;
 	PVShared   *shared;
-	VacDeadItems *dead_items;
+	TIDStore	*dead_items;
 	PVIndStats *indstats;
 	BufferUsage *buffer_usage;
 	WalUsage   *wal_usage;
+	void		*area_space;
+	dsa_area	*dead_items_dsa;
 	bool	   *will_parallel_vacuum;
 	Size		est_indstats_len;
 	Size		est_shared_len;
-	Size		est_dead_items_len;
+	Size		dsa_minsize = dsa_minimum_size();
 	int			nindexes_mwm = 0;
 	int			parallel_workers = 0;
 	int			querylen;
@@ -283,9 +290,8 @@ parallel_vacuum_init(Relation rel, Relation *indrels, int nindexes,
 	shm_toc_estimate_chunk(&pcxt->estimator, est_shared_len);
 	shm_toc_estimate_keys(&pcxt->estimator, 1);
 
-	/* Estimate size for dead_items -- PARALLEL_VACUUM_KEY_DEAD_ITEMS */
-	est_dead_items_len = vac_max_items_to_alloc_size(max_items);
-	shm_toc_estimate_chunk(&pcxt->estimator, est_dead_items_len);
+	/* Estimate size for dead tuple DSA -- PARALLEL_VACUUM_KEY_DSA */
+	shm_toc_estimate_chunk(&pcxt->estimator, dsa_minsize);
 	shm_toc_estimate_keys(&pcxt->estimator, 1);
 
 	/*
@@ -351,6 +357,16 @@ parallel_vacuum_init(Relation rel, Relation *indrels, int nindexes,
 	shm_toc_insert(pcxt->toc, PARALLEL_VACUUM_KEY_INDEX_STATS, indstats);
 	pvs->indstats = indstats;
 
+	/* Prepare DSA space for dead items */
+	area_space = shm_toc_allocate(pcxt->toc, dsa_minsize);
+	shm_toc_insert(pcxt->toc, PARALLEL_VACUUM_KEY_DSA, area_space);
+	dead_items_dsa = dsa_create_in_place(area_space, dsa_minsize,
+										 LWTRANCHE_PARALLEL_VACUUM_DSA,
+										 pcxt->seg);
+	dead_items = tidstore_create(vac_work_mem, dead_items_dsa);
+	pvs->dead_items = dead_items;
+	pvs->dead_items_area = dead_items_dsa;
+
 	/* Prepare shared information */
 	shared = (PVShared *) shm_toc_allocate(pcxt->toc, est_shared_len);
 	MemSet(shared, 0, est_shared_len);
@@ -360,6 +376,7 @@ parallel_vacuum_init(Relation rel, Relation *indrels, int nindexes,
 		(nindexes_mwm > 0) ?
 		maintenance_work_mem / Min(parallel_workers, nindexes_mwm) :
 		maintenance_work_mem;
+	shared->dead_items_handle = tidstore_get_handle(dead_items);
 
 	pg_atomic_init_u32(&(shared->cost_balance), 0);
 	pg_atomic_init_u32(&(shared->active_nworkers), 0);
@@ -368,15 +385,6 @@ parallel_vacuum_init(Relation rel, Relation *indrels, int nindexes,
 	shm_toc_insert(pcxt->toc, PARALLEL_VACUUM_KEY_SHARED, shared);
 	pvs->shared = shared;
 
-	/* Prepare the dead_items space */
-	dead_items = (VacDeadItems *) shm_toc_allocate(pcxt->toc,
-												   est_dead_items_len);
-	dead_items->max_items = max_items;
-	dead_items->num_items = 0;
-	MemSet(dead_items->items, 0, sizeof(ItemPointerData) * max_items);
-	shm_toc_insert(pcxt->toc, PARALLEL_VACUUM_KEY_DEAD_ITEMS, dead_items);
-	pvs->dead_items = dead_items;
-
 	/*
 	 * Allocate space for each worker's BufferUsage and WalUsage; no need to
 	 * initialize
@@ -434,6 +442,9 @@ parallel_vacuum_end(ParallelVacuumState *pvs, IndexBulkDeleteResult **istats)
 			istats[i] = NULL;
 	}
 
+	tidstore_free(pvs->dead_items);
+	dsa_detach(pvs->dead_items_area);
+
 	DestroyParallelContext(pvs->pcxt);
 	ExitParallelMode();
 
@@ -442,7 +453,7 @@ parallel_vacuum_end(ParallelVacuumState *pvs, IndexBulkDeleteResult **istats)
 }
 
 /* Returns the dead items space */
-VacDeadItems *
+TIDStore *
 parallel_vacuum_get_dead_items(ParallelVacuumState *pvs)
 {
 	return pvs->dead_items;
@@ -940,7 +951,9 @@ parallel_vacuum_main(dsm_segment *seg, shm_toc *toc)
 	Relation   *indrels;
 	PVIndStats *indstats;
 	PVShared   *shared;
-	VacDeadItems *dead_items;
+	TIDStore	*dead_items;
+	void		*area_space;
+	dsa_area	*dead_items_area;
 	BufferUsage *buffer_usage;
 	WalUsage   *wal_usage;
 	int			nindexes;
@@ -984,10 +997,10 @@ parallel_vacuum_main(dsm_segment *seg, shm_toc *toc)
 											 PARALLEL_VACUUM_KEY_INDEX_STATS,
 											 false);
 
-	/* Set dead_items space */
-	dead_items = (VacDeadItems *) shm_toc_lookup(toc,
-												 PARALLEL_VACUUM_KEY_DEAD_ITEMS,
-												 false);
+	/* Set dead items */
+	area_space = shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_DSA, false);
+	dead_items_area = dsa_attach_in_place(area_space, seg);
+	dead_items = tidstore_attach(dead_items_area, shared->dead_items_handle);
 
 	/* Set cost-based vacuum delay */
 	VacuumCostActive = (VacuumCostDelay > 0);
@@ -1033,6 +1046,9 @@ parallel_vacuum_main(dsm_segment *seg, shm_toc *toc)
 	InstrEndParallelQuery(&buffer_usage[ParallelWorkerNumber],
 						  &wal_usage[ParallelWorkerNumber]);
 
+	tidstore_detach(pvs.dead_items);
+	dsa_detach(dead_items_area);
+
 	/* Pop the error context stack */
 	error_context_stack = errcallback.previous;
 
diff --git a/src/backend/storage/lmgr/lwlock.c b/src/backend/storage/lmgr/lwlock.c
index 528b2e9643..ea8cf6283b 100644
--- a/src/backend/storage/lmgr/lwlock.c
+++ b/src/backend/storage/lmgr/lwlock.c
@@ -186,6 +186,8 @@ static const char *const BuiltinTrancheNames[] = {
 	"PgStatsHash",
 	/* LWTRANCHE_PGSTATS_DATA: */
 	"PgStatsData",
+	/* LWTRANCHE_PARALLEL_VACUUM_DSA: */
+	"ParallelVacuumDSA",
 };
 
 StaticAssertDecl(lengthof(BuiltinTrancheNames) ==
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index 1bf14eec66..5d9808977e 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -2280,7 +2280,7 @@ struct config_int ConfigureNamesInt[] =
 			GUC_UNIT_KB
 		},
 		&maintenance_work_mem,
-		65536, 1024, MAX_KILOBYTES,
+		65536, 2048, MAX_KILOBYTES,
 		NULL, NULL, NULL
 	},
 
diff --git a/src/include/access/tidstore.h b/src/include/access/tidstore.h
new file mode 100644
index 0000000000..4a7ab3f5a8
--- /dev/null
+++ b/src/include/access/tidstore.h
@@ -0,0 +1,49 @@
+/*-------------------------------------------------------------------------
+ *
+ * tidstore.h
+ *	  TID storage.
+ *
+ *
+ * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/access/tidstore.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef TIDSTORE_H
+#define TIDSTORE_H
+
+#include "lib/radixtree.h"
+#include "storage/itemptr.h"
+
+typedef dsa_pointer tidstore_handle;
+
+typedef struct TIDStore TIDStore;
+typedef struct TIDStoreIter TIDStoreIter;
+
+typedef struct TIDStoreIterResult
+{
+	BlockNumber		blkno;
+	OffsetNumber	offsets[MaxOffsetNumber]; /* XXX: usually don't use up */
+	int				num_offsets;
+} TIDStoreIterResult;
+
+extern TIDStore *tidstore_create(uint64 max_bytes, dsa_area *dsa);
+extern TIDStore *tidstore_attach(dsa_area *dsa, dsa_pointer handle);
+extern void tidstore_detach(TIDStore *ts);
+extern void tidstore_free(TIDStore *ts);
+extern void tidstore_reset(TIDStore *ts);
+extern void tidstore_add_tids(TIDStore *ts, BlockNumber blkno, OffsetNumber *offsets,
+							  int num_offsets);
+extern bool tidstore_lookup_tid(TIDStore *ts, ItemPointer tid);
+extern TIDStoreIter * tidstore_begin_iterate(TIDStore *ts);
+extern TIDStoreIterResult *tidstore_iterate_next(TIDStoreIter *iter);
+extern uint64 tidstore_num_tids(TIDStore *ts);
+extern bool tidstore_is_full(TIDStore *ts);
+extern uint64 tidstore_max_memory(TIDStore *ts);
+extern uint64 tidstore_memory_usage(TIDStore *ts);
+extern tidstore_handle tidstore_get_handle(TIDStore *ts);
+
+#endif		/* TIDSTORE_H */
+
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
index a28938caf4..75d540d315 100644
--- a/src/include/commands/progress.h
+++ b/src/include/commands/progress.h
@@ -23,8 +23,8 @@
 #define PROGRESS_VACUUM_HEAP_BLKS_SCANNED		2
 #define PROGRESS_VACUUM_HEAP_BLKS_VACUUMED		3
 #define PROGRESS_VACUUM_NUM_INDEX_VACUUMS		4
-#define PROGRESS_VACUUM_MAX_DEAD_TUPLES			5
-#define PROGRESS_VACUUM_NUM_DEAD_TUPLES			6
+#define PROGRESS_VACUUM_MAX_DEAD_TUPLE_BYTES	5
+#define PROGRESS_VACUUM_DEAD_TUPLE_BYTES		6
 
 /* Phases of vacuum (as advertised via PROGRESS_VACUUM_PHASE) */
 #define PROGRESS_VACUUM_PHASE_SCAN_HEAP			1
diff --git a/src/include/commands/vacuum.h b/src/include/commands/vacuum.h
index 4e4bc26a8b..afe61c21fd 100644
--- a/src/include/commands/vacuum.h
+++ b/src/include/commands/vacuum.h
@@ -17,6 +17,7 @@
 #include "access/htup.h"
 #include "access/genam.h"
 #include "access/parallel.h"
+#include "access/tidstore.h"
 #include "catalog/pg_class.h"
 #include "catalog/pg_statistic.h"
 #include "catalog/pg_type.h"
@@ -235,21 +236,6 @@ typedef struct VacuumParams
 	int			nworkers;
 } VacuumParams;
 
-/*
- * VacDeadItems stores TIDs whose index tuples are deleted by index vacuuming.
- */
-typedef struct VacDeadItems
-{
-	int			max_items;		/* # slots allocated in array */
-	int			num_items;		/* current # of entries */
-
-	/* Sorted array of TIDs to delete from indexes */
-	ItemPointerData items[FLEXIBLE_ARRAY_MEMBER];
-} VacDeadItems;
-
-#define MAXDEADITEMS(avail_mem) \
-	(((avail_mem) - offsetof(VacDeadItems, items)) / sizeof(ItemPointerData))
-
 /* GUC parameters */
 extern PGDLLIMPORT int default_statistics_target;	/* PGDLLIMPORT for PostGIS */
 extern PGDLLIMPORT int vacuum_freeze_min_age;
@@ -302,18 +288,17 @@ extern Relation vacuum_open_relation(Oid relid, RangeVar *relation,
 									 LOCKMODE lmode);
 extern IndexBulkDeleteResult *vac_bulkdel_one_index(IndexVacuumInfo *ivinfo,
 													IndexBulkDeleteResult *istat,
-													VacDeadItems *dead_items);
+													TIDStore *dead_items);
 extern IndexBulkDeleteResult *vac_cleanup_one_index(IndexVacuumInfo *ivinfo,
 													IndexBulkDeleteResult *istat);
-extern Size vac_max_items_to_alloc_size(int max_items);
 
 /* in commands/vacuumparallel.c */
 extern ParallelVacuumState *parallel_vacuum_init(Relation rel, Relation *indrels,
 												 int nindexes, int nrequested_workers,
-												 int max_items, int elevel,
-												 BufferAccessStrategy bstrategy);
+												 int vac_work_mem,
+												 int elevel, BufferAccessStrategy bstrategy);
 extern void parallel_vacuum_end(ParallelVacuumState *pvs, IndexBulkDeleteResult **istats);
-extern VacDeadItems *parallel_vacuum_get_dead_items(ParallelVacuumState *pvs);
+extern TIDStore *parallel_vacuum_get_dead_items(ParallelVacuumState *pvs);
 extern void parallel_vacuum_bulkdel_all_indexes(ParallelVacuumState *pvs,
 												long num_table_tuples,
 												int num_index_scans);
diff --git a/src/include/storage/lwlock.h b/src/include/storage/lwlock.h
index dd818e16ab..f1e0bcede5 100644
--- a/src/include/storage/lwlock.h
+++ b/src/include/storage/lwlock.h
@@ -204,6 +204,7 @@ typedef enum BuiltinTrancheIds
 	LWTRANCHE_PGSTATS_DSA,
 	LWTRANCHE_PGSTATS_HASH,
 	LWTRANCHE_PGSTATS_DATA,
+	LWTRANCHE_PARALLEL_VACUUM_DSA,
 	LWTRANCHE_FIRST_USER_DEFINED
 }			BuiltinTrancheIds;
 
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index fb9f936d43..0c49354f04 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -2020,8 +2020,8 @@ pg_stat_progress_vacuum| SELECT s.pid,
     s.param3 AS heap_blks_scanned,
     s.param4 AS heap_blks_vacuumed,
     s.param5 AS index_vacuum_count,
-    s.param6 AS max_dead_tuples,
-    s.param7 AS num_dead_tuples
+    s.param6 AS max_dead_tuple_bytes,
+    s.param7 AS dead_tuple_bytes
    FROM (pg_stat_get_progress_info('VACUUM'::text) s(pid, datid, relid, param1, param2, param3, param4, param5, param6, param7, param8, param9, param10, param11, param12, param13, param14, param15, param16, param17, param18, param19, param20)
      LEFT JOIN pg_database d ON ((s.datid = d.oid)));
 pg_stat_recovery_prefetch| SELECT s.stats_reset,
-- 
2.31.1