From 3878f7c8257999ed6b6e5d93996f5abb43ba0b2f Mon Sep 17 00:00:00 2001
From: Hannu Krosing <hannuk@google.com>
Date: Sun, 23 Aug 2026 20:13:04 +0000
Subject: [PATCH v2 4/8] Support Direct TOAST in logical decoding, replication,
 and online REPACK

Add logical replication and decoding support for Direct TOAST tuples:
- reorderbuffer.c: Support direct TOAST chunk reconstruction by keying the toast
  hash table on tuple TID (via a union key) when chunk_id is InvalidOid. Reconstruct
  hierarchical direct TOAST DAGs bottom-up from the WAL change stream.
- decode.c: Record the tuple's physical TID in decode_heap() when processing toast inserts.
- proto.c & pgoutput.c: Correctly handle unchanged direct TOAST attributes in the logical
  replication output plugin.
- Add isolation tests using injection points (repack_direct_toast.spec).
---
 src/backend/commands/repack.c                 |  22 +-
 src/backend/replication/logical/decode.c      |   7 +-
 src/backend/replication/logical/proto.c       |   2 +-
 .../replication/logical/reorderbuffer.c       | 188 +++++++++++++++---
 src/backend/replication/pgoutput/pgoutput.c   |   4 +-
 src/backend/replication/pgrepack/pgrepack.c   |   2 +-
 src/test/modules/injection_points/Makefile    |   1 +
 .../expected/repack_direct_toast.out          | 126 ++++++++++++
 .../specs/repack_direct_toast.spec            | 180 +++++++++++++++++
 9 files changed, 499 insertions(+), 33 deletions(-)
 create mode 100644 src/test/modules/injection_points/expected/repack_direct_toast.out
 create mode 100644 src/test/modules/injection_points/specs/repack_direct_toast.spec

diff --git a/src/backend/commands/repack.c b/src/backend/commands/repack.c
index 477c86b2ba6..d77cd55d332 100644
--- a/src/backend/commands/repack.c
+++ b/src/backend/commands/repack.c
@@ -621,6 +621,26 @@ cluster_rel(RepackCommand cmd, Relation OldHeap, Oid indexOid,
 		goto out;
 	}
 
+	if (OldHeap->rd_rel->relkind == RELKIND_TOASTVALUE)
+	{
+		/*
+		 * If this TOAST table supports Direct TOAST (has chunk_tids column),
+		 * disallow VACUUM FULL, CLUSTER, or REPACK directly on the TOAST
+		 * table. Rebuilding the TOAST table independently would invalidate the
+		 * physical TIDs stored in the parent relation's tuples.
+		 */
+		if (OldHeap->rd_att->natts >= 4 &&
+			strcmp(NameStr(TupleDescAttr(OldHeap->rd_att, 3)->attname), "chunk_tids") == 0)
+		{
+			ereport(ERROR,
+					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+					 errmsg("cannot %s direct TOAST table directly",
+							RepackCommandAsString(cmd)),
+					 errhint("Execute %s on the parent table instead.",
+							 RepackCommandAsString(cmd))));
+		}
+	}
+
 	Assert(OldHeap->rd_rel->relkind == RELKIND_RELATION ||
 		   OldHeap->rd_rel->relkind == RELKIND_MATVIEW ||
 		   OldHeap->rd_rel->relkind == RELKIND_TOASTVALUE);
@@ -2874,7 +2894,7 @@ adjust_toast_pointers(Relation relation, TupleTableSlot *dest, TupleTableSlot *s
 		slot_getsomeattrs(dest, i + 1);
 
 		varlena_dst = (varlena *) DatumGetPointer(dest->tts_values[i]);
-		if (!VARATT_IS_EXTERNAL_ONDISK(varlena_dst))
+		if (!VARATT_IS_EXTERNAL_ONDISK(varlena_dst) && !VARATT_IS_EXTERNAL_DIRECT(varlena_dst))
 			continue;
 		slot_getsomeattrs(src, i + 1);
 
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index c944be4ac83..fe801cf27fb 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -921,6 +921,7 @@ DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 	xl_heap_insert *xlrec;
 	ReorderBufferChange *change;
 	RelFileLocator target_locator;
+	BlockNumber blknum;
 
 	xlrec = (xl_heap_insert *) XLogRecGetData(r);
 
@@ -932,7 +933,7 @@ DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 		return;
 
 	/* only interested in our database */
-	XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
+	XLogRecGetBlockTag(r, 0, &target_locator, NULL, &blknum);
 	if (target_locator.dbOid != ctx->slot->data.database)
 		return;
 
@@ -957,6 +958,10 @@ DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
 	DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple);
 
+	/* Set the TID for toast relation inserts, needed for direct toast */
+	if (xlrec->flags & XLH_INSERT_ON_TOAST_RELATION)
+		ItemPointerSet(&change->data.tp.newtuple->t_self, blknum, xlrec->offnum);
+
 	change->data.tp.clear_toast_afterwards = true;
 
 	ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index 86ad97cd937..50f29523499 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -812,7 +812,7 @@ logicalrep_write_tuple(StringInfo out, Relation rel, TupleTableSlot *slot,
 			continue;
 		}
 
-		if (att->attlen == -1 && VARATT_IS_EXTERNAL_ONDISK(DatumGetPointer(values[i])))
+		if (att->attlen == -1 && VARATT_IS_EXTERNAL_ONDISK_OR_DIRECT(DatumGetPointer(values[i])))
 		{
 			/*
 			 * Unchanged toasted datum.  (Note that we don't promise to detect
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 900864afc6d..5b9dfb3e379 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -110,6 +110,9 @@
 #include "storage/sinval.h"
 #include "utils/builtins.h"
 #include "utils/inval.h"
+#include "utils/array.h"
+#include "catalog/pg_type.h"
+#include "utils/lsyscache.h"
 #include "utils/memutils.h"
 #include "utils/rel.h"
 #include "utils/relfilenumbermap.h"
@@ -174,10 +177,21 @@ typedef struct ReorderBufferIterTXNState
 	ReorderBufferIterTXNEntry entries[FLEXIBLE_ARRAY_MEMBER];
 } ReorderBufferIterTXNState;
 
+typedef struct ReorderBufferToastKey
+{
+	bool		is_direct;
+	union
+	{
+		Oid			chunk_id;
+		ItemPointerData tid;
+	}			u;
+} ReorderBufferToastKey;
+
 /* toast datastructures */
 typedef struct ReorderBufferToastEnt
 {
-	Oid			chunk_id;		/* toast_table.chunk_id */
+	ReorderBufferToastKey key;	/* toast_table.chunk_id or chunk TID */
+	Oid			chunk_id;		/* toast_table.chunk_id (kept for compatibility) */
 	int32		last_chunk_seq; /* toast_table.chunk_seq of the last chunk we
 								 * have seen */
 	Size		num_chunks;		/* number of chunks we've already seen */
@@ -5010,7 +5024,7 @@ ReorderBufferToastInitHash(ReorderBuffer *rb, ReorderBufferTXN *txn)
 
 	Assert(txn->toast_hash == NULL);
 
-	hash_ctl.keysize = sizeof(Oid);
+	hash_ctl.keysize = sizeof(ReorderBufferToastKey);
 	hash_ctl.entrysize = sizeof(ReorderBufferToastEnt);
 	hash_ctl.hcxt = rb->context;
 	txn->toast_hash = hash_create("ReorderBufferToastHash", 5, &hash_ctl,
@@ -5036,6 +5050,10 @@ ReorderBufferToastAppendChunk(ReorderBuffer *rb, ReorderBufferTXN *txn,
 	TupleDesc	desc = RelationGetDescr(relation);
 	Oid			chunk_id;
 	int32		chunk_seq;
+	ReorderBufferToastKey key;
+	bool		is_direct = false;
+	Datum		chunk_tids_datum;
+	bool		chunk_tids_isnull = true;
 
 	if (txn->toast_hash == NULL)
 		ReorderBufferToastInitHash(rb, txn);
@@ -5044,46 +5062,139 @@ ReorderBufferToastAppendChunk(ReorderBuffer *rb, ReorderBufferTXN *txn,
 
 	newtup = change->data.tp.newtuple;
 	chunk_id = DatumGetObjectId(fastgetattr(newtup, 1, desc, &isnull));
-	Assert(!isnull);
+	if (isnull)
+	{
+		is_direct = true;
+		memset(&key, 0, sizeof(key));
+		key.is_direct = true;
+		key.u.tid = newtup->t_self;
+		if (!ItemPointerIsValid(&key.u.tid))
+			elog(ERROR, "invalid TID for direct toast chunk");
+	}
+	else
+	{
+		memset(&key, 0, sizeof(key));
+		key.is_direct = false;
+		key.u.chunk_id = chunk_id;
+	}
+
 	chunk_seq = DatumGetInt32(fastgetattr(newtup, 2, desc, &isnull));
 	Assert(!isnull);
 
+	if (desc->natts >= 4)
+	{
+		chunk_tids_datum = fastgetattr(newtup, 4, desc, &chunk_tids_isnull);
+	}
+
 	ent = (ReorderBufferToastEnt *)
-		hash_search(txn->toast_hash, &chunk_id, HASH_ENTER, &found);
+		hash_search(txn->toast_hash, &key, HASH_ENTER, &found);
+
+	chunk = DatumGetPointer(fastgetattr(newtup, 3, desc, &isnull));
+	if (!isnull)
+	{
+		/* calculate size so we can allocate the right size at once later */
+		if (!VARATT_IS_EXTENDED(chunk))
+			chunksize = VARSIZE(chunk) - VARHDRSZ;
+		else if (VARATT_IS_SHORT(chunk))
+			/* could happen due to heap_form_tuple doing its thing */
+			chunksize = VARSIZE_SHORT(chunk) - VARHDRSZ_SHORT;
+		else
+			elog(ERROR, "unexpected type of toast chunk");
+	}
+	else
+	{
+		chunksize = 0;
+	}
 
 	if (!found)
 	{
-		Assert(ent->chunk_id == chunk_id);
+		ent->key = key;
+		ent->chunk_id = is_direct ? InvalidOid : chunk_id;
 		ent->num_chunks = 0;
 		ent->last_chunk_seq = 0;
 		ent->size = 0;
 		ent->reconstructed = NULL;
 		dlist_init(&ent->chunks);
 
-		if (chunk_seq != 0)
+		if (!is_direct && chunk_seq != 0)
 			elog(ERROR, "got sequence entry %d for toast chunk %u instead of seq 0",
 				 chunk_seq, chunk_id);
 	}
-	else if (found && chunk_seq != ent->last_chunk_seq + 1)
-		elog(ERROR, "got sequence entry %d for toast chunk %u instead of seq %d",
-			 chunk_seq, chunk_id, ent->last_chunk_seq + 1);
+	else if (found && !is_direct)
+	{
+		if (chunk_seq != ent->last_chunk_seq + 1)
+			elog(ERROR, "got sequence entry %d for toast chunk %u instead of seq %d",
+				 chunk_seq, chunk_id, ent->last_chunk_seq + 1);
+	}
+	else if (found && is_direct)
+	{
+		elog(ERROR, "duplicate TID in direct toast hash");
+	}
 
-	chunk = DatumGetPointer(fastgetattr(newtup, 3, desc, &isnull));
-	Assert(!isnull);
+	/* Group previous chunks if this is a node with chunk_tids */
+	if (is_direct && !chunk_tids_isnull)
+	{
+		ArrayType  *arr = DatumGetArrayTypeP(chunk_tids_datum);
+		Oid			eltype = ARR_ELEMTYPE(arr);
+		int16		typlen;
+		bool		typbyval;
+		char		typalign;
+		Datum	   *elems;
+		bool	   *nulls;
+		int			nelems;
+		int			i;
+
+		Assert(eltype == TIDOID);
+		get_typlenbyvalalign(eltype, &typlen, &typbyval, &typalign);
+		deconstruct_array(arr, eltype, typlen, typbyval, typalign,
+						  &elems, &nulls, &nelems);
+
+		for (i = 0; i < nelems; i++)
+		{
+			ItemPointer tid = DatumGetItemPointer(elems[i]);
+			ReorderBufferToastKey prev_key;
+			ReorderBufferToastEnt *ent_prev;
+			dlist_mutable_iter miter;
 
-	/* calculate size so we can allocate the right size at once later */
-	if (!VARATT_IS_EXTENDED(chunk))
-		chunksize = VARSIZE(chunk) - VARHDRSZ;
-	else if (VARATT_IS_SHORT(chunk))
-		/* could happen due to heap_form_tuple doing its thing */
-		chunksize = VARSIZE_SHORT(chunk) - VARHDRSZ_SHORT;
-	else
-		elog(ERROR, "unexpected type of toast chunk");
+			if (ItemPointerEquals(tid, &key.u.tid))
+				continue;
+
+			memset(&prev_key, 0, sizeof(prev_key));
+			prev_key.is_direct = true;
+			prev_key.u.tid = *tid;
+
+			ent_prev = (ReorderBufferToastEnt *)
+				hash_search(txn->toast_hash, &prev_key, HASH_FIND, NULL);
+
+			if (ent_prev == NULL)
+				elog(ERROR, "could not find previous direct toast chunk");
+
+			/* Move chunks to the current entry */
+			dlist_foreach_modify(miter, &ent_prev->chunks)
+			{
+				ReorderBufferChange *c = dlist_container(ReorderBufferChange, node, miter.cur);
+				dlist_delete(miter.cur);
+				dlist_push_tail(&ent->chunks, &c->node);
+			}
+
+			ent->size += ent_prev->size;
+			ent->num_chunks += ent_prev->num_chunks;
+
+			/* Remove from hash */
+			hash_search(txn->toast_hash, &prev_key, HASH_REMOVE, NULL);
+		}
+
+		pfree(elems);
+		pfree(nulls);
+	}
 
-	ent->size += chunksize;
 	ent->last_chunk_seq = chunk_seq;
-	ent->num_chunks++;
-	dlist_push_tail(&ent->chunks, &change->node);
+	if (chunksize > 0)
+	{
+		ent->size += chunksize;
+		ent->num_chunks++;
+		dlist_push_tail(&ent->chunks, &change->node);
+	}
 }
 
 /*
@@ -5174,6 +5285,10 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
 		varlena    *reconstructed;
 		dlist_iter	it;
 		Size		data_done = 0;
+		ReorderBufferToastKey key;
+		int32		rawsize;
+		uint32		extsize;
+		bool		is_compressed;
 
 		if (attr->attisdropped)
 			continue;
@@ -5193,14 +5308,33 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
 		if (!VARATT_IS_EXTERNAL(varlena_pointer))
 			continue;
 
-		VARATT_EXTERNAL_GET_POINTER(toast_pointer, varlena_pointer);
+		memset(&key, 0, sizeof(key));
+		if (VARATT_IS_EXTERNAL_DIRECT(varlena_pointer))
+		{
+			varatt_direct toast_pointer_direct;
+			VARATT_EXTERNAL_GET_POINTER_DIRECT(toast_pointer_direct, varlena_pointer);
+			key.is_direct = true;
+			key.u.tid = toast_pointer_direct.va_tid;
+			rawsize = toast_pointer_direct.va_rawsize;
+			extsize = VARATT_DIRECT_GET_EXTSIZE(toast_pointer_direct);
+			is_compressed = VARATT_DIRECT_IS_COMPRESSED(toast_pointer_direct);
+		}
+		else
+		{
+			VARATT_EXTERNAL_GET_POINTER(toast_pointer, varlena_pointer);
+			key.is_direct = false;
+			key.u.chunk_id = toast_pointer.va_valueid;
+			rawsize = toast_pointer.va_rawsize;
+			extsize = VARATT_EXTERNAL_GET_EXTSIZE(toast_pointer);
+			is_compressed = VARATT_EXTERNAL_IS_COMPRESSED(toast_pointer);
+		}
 
 		/*
 		 * Check whether the toast tuple changed, replace if so.
 		 */
 		ent = (ReorderBufferToastEnt *)
 			hash_search(txn->toast_hash,
-						&toast_pointer.va_valueid,
+						&key,
 						HASH_FIND,
 						NULL);
 		if (ent == NULL)
@@ -5211,7 +5345,7 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
 
 		free[natt] = true;
 
-		reconstructed = palloc0(toast_pointer.va_rawsize);
+		reconstructed = palloc0(rawsize);
 
 		ent->reconstructed = reconstructed;
 
@@ -5236,10 +5370,10 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
 				   VARSIZE(chunk) - VARHDRSZ);
 			data_done += VARSIZE(chunk) - VARHDRSZ;
 		}
-		Assert(data_done == VARATT_EXTERNAL_GET_EXTSIZE(toast_pointer));
+		Assert(data_done == extsize);
 
 		/* make sure its marked as compressed or not */
-		if (VARATT_EXTERNAL_IS_COMPRESSED(toast_pointer))
+		if (is_compressed)
 			SET_VARSIZE_COMPRESSED(reconstructed, data_done + VARHDRSZ);
 		else
 			SET_VARSIZE(reconstructed, data_done + VARHDRSZ);
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 484ffbe2cee..3703c47148c 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -1402,8 +1402,8 @@ pgoutput_row_filter(Relation relation, TupleTableSlot *old_slot,
 		 * VARTAG_INDIRECT. See ReorderBufferToastReplace.
 		 */
 		if (att->attlen == -1 &&
-			VARATT_IS_EXTERNAL_ONDISK(DatumGetPointer(new_slot->tts_values[i])) &&
-			!VARATT_IS_EXTERNAL_ONDISK(DatumGetPointer(old_slot->tts_values[i])))
+			VARATT_IS_EXTERNAL_ONDISK_OR_DIRECT(DatumGetPointer(new_slot->tts_values[i])) &&
+			!VARATT_IS_EXTERNAL_ONDISK_OR_DIRECT(DatumGetPointer(old_slot->tts_values[i])))
 		{
 			if (!tmp_new_slot)
 			{
diff --git a/src/backend/replication/pgrepack/pgrepack.c b/src/backend/replication/pgrepack/pgrepack.c
index a1cbc8db68a..38fb231ad98 100644
--- a/src/backend/replication/pgrepack/pgrepack.c
+++ b/src/backend/replication/pgrepack/pgrepack.c
@@ -272,7 +272,7 @@ repack_store_change(LogicalDecodingContext *ctx, Relation relation,
 				 * We get here if the table has external values but only
 				 * in-line values are being updated now.
 				 */
-				Assert(VARATT_IS_EXTERNAL_ONDISK(varlen));
+				Assert(VARATT_IS_EXTERNAL_ONDISK(varlen) || VARATT_IS_EXTERNAL_DIRECT(varlen));
 			}
 		}
 
diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile
index 3b136adf126..bc04c09d75e 100644
--- a/src/test/modules/injection_points/Makefile
+++ b/src/test/modules/injection_points/Makefile
@@ -19,6 +19,7 @@ ISOLATION = basic \
 	    repack_temporal \
 	    repack_temporal_multirange \
 	    repack_toast \
+	    repack_direct_toast \
 	    ri_fastpath_reindex \
 	    syscache-update-pruned \
 	    wait_cleanup \
diff --git a/src/test/modules/injection_points/expected/repack_direct_toast.out b/src/test/modules/injection_points/expected/repack_direct_toast.out
new file mode 100644
index 00000000000..aa73c3a24cc
--- /dev/null
+++ b/src/test/modules/injection_points/expected/repack_direct_toast.out
@@ -0,0 +1,126 @@
+Parsed test spec with 2 sessions
+
+starting permutation: s1_wait_before_lock s2_updates s2_check s2_wakeup_before_lock s1_check
+injection_points_attach
+-----------------------
+                       
+(1 row)
+
+step s1_wait_before_lock: 
+	REPACK (CONCURRENTLY) repack_direct_toast;
+ <waiting ...>
+step s2_updates: 
+	DELETE FROM repack_direct_toast WHERE i=1;
+	INSERT INTO repack_direct_toast(i, j, k) VALUES (1, gen_external(), gen_compressible(1));
+
+	-- existing toast data unchanged.  (This covers the case where we
+	-- adjust the toast pointer.)
+	UPDATE repack_direct_toast SET i=i+300 where i % 10 = 2 RETURNING OLD.i, NEW.i;
+
+	-- "j" is here an external indirect, written to the file separately.
+	UPDATE repack_direct_toast SET j=gen_external() where i % 10 = 3 RETURNING OLD.i, NEW.i;
+
+	-- the updated value of "j" is compressed.
+	UPDATE repack_direct_toast SET j=gen_compressible(1), k=k||'' where i % 10 = 4 RETURNING i;
+
+	-- the updated value of "j" is compressed externally.
+	UPDATE repack_direct_toast SET j=gen_compressible_external(2) where i % 10 = 5 RETURNING i;
+
+	-- the updated value of "j" stays inline.
+	UPDATE repack_direct_toast SET j=gen_inline(), k=repeat(k,5) where i % 10 = 6 RETURNING i;
+
+	-- updated value of "j" is a short varlena; "k" is written separately.
+	UPDATE repack_direct_toast SET j=gen_short(), k=gen_external() where i % 10 = 7 RETURNING i;
+
+ i|  i
+--+---
+ 2|302
+12|312
+(2 rows)
+
+ i| i
+--+--
+ 3| 3
+13|13
+(2 rows)
+
+ i
+--
+ 4
+14
+(2 rows)
+
+ i
+--
+ 5
+15
+(2 rows)
+
+ i
+--
+ 6
+16
+(2 rows)
+
+ i
+--
+ 7
+17
+(2 rows)
+
+step s2_check: 
+	INSERT INTO relfilenodes(node)
+	SELECT c2.relfilenode
+	FROM pg_class c1 JOIN pg_class c2 ON c2.oid = c1.oid OR c2.oid = c1.reltoastrelid
+	WHERE c1.relname='repack_direct_toast';
+
+	INSERT INTO data_s2(i, j, j_toast, k, k_toast)
+	SELECT i, j, COALESCE(pg_column_toast_chunk_id(j), 0) AS j_toast,
+	k, COALESCE(pg_column_toast_chunk_id(k), 0) AS k_toast
+	FROM repack_direct_toast;
+
+step s2_wakeup_before_lock: 
+	SELECT injection_points_wakeup('repack-concurrently-before-lock');
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+step s1_wait_before_lock: <... completed>
+step s1_check: 
+	INSERT INTO relfilenodes(node)
+	SELECT c2.relfilenode
+	FROM pg_class c1 JOIN pg_class c2 ON c2.oid = c1.oid OR c2.oid = c1.reltoastrelid
+	WHERE c1.relname='repack_direct_toast';
+
+	SELECT count(DISTINCT node) FROM relfilenodes;
+
+	INSERT INTO data_s1(i, j, j_toast, k, k_toast)
+	SELECT i,
+	j, COALESCE(pg_column_toast_chunk_id(j), 0) AS j_toast,
+	k, COALESCE(pg_column_toast_chunk_id(k), 0) AS k_toast
+	FROM repack_direct_toast;
+
+	-- this should be empty
+	SELECT d1.i, substring(d1.j FOR 12) AS d1_j, substring(d1.k FOR 12) AS d1_k,
+		d2.i, substring(d2.j FOR 12) AS d2_j, substring(d2.k FOR 12) AS d2_k,
+		d1.j_toast as d1_j_tst, d2.j_toast as d2_j_tst,
+		d1.k_toast as d1_k_tst, d2.k_toast AS d2_k_tst
+	FROM data_s1 d1 FULL JOIN data_s2 d2 USING (i, j, k)
+	WHERE d1.i ISNULL OR d2.i ISNULL;
+
+count
+-----
+    4
+(1 row)
+
+i|d1_j|d1_k|i|d2_j|d2_k|d1_j_tst|d2_j_tst|d1_k_tst|d2_k_tst
+-+----+----+-+----+----+--------+--------+--------+--------
+(0 rows)
+
+injection_points_detach
+-----------------------
+                       
+(1 row)
+
diff --git a/src/test/modules/injection_points/specs/repack_direct_toast.spec b/src/test/modules/injection_points/specs/repack_direct_toast.spec
new file mode 100644
index 00000000000..c911349eb1a
--- /dev/null
+++ b/src/test/modules/injection_points/specs/repack_direct_toast.spec
@@ -0,0 +1,180 @@
+# REPACK (CONCURRENTLY) with Direct TOAST;
+#
+# Test handling of Direct TOAST. At the same time, no tuplesort.
+setup
+{
+
+	CREATE EXTENSION IF NOT EXISTS injection_points;
+
+	-- Generate text consisting of repeated strings so that it can be
+	-- compressed easily.
+	CREATE FUNCTION gen_compressible(seed int)
+	RETURNS text
+	LANGUAGE sql IMMUTABLE as $$
+		SELECT repeat(md5((seed * 1000)::text), 50);
+	$$;
+
+	-- Like above, but too big even after compression.
+	CREATE FUNCTION gen_compressible_external(seed int)
+	RETURNS text
+	LANGUAGE sql IMMUTABLE as $$
+		SELECT repeat(md5((seed * 1000)::text), 10000);
+	$$;
+
+	-- Generate a string of random characters that is not likely to be
+	-- compressed, but is big enough to be stored externally.
+	CREATE FUNCTION gen_external()
+	RETURNS text
+	LANGUAGE sql as $$
+		SELECT string_agg(chr(65 + trunc(25 * random())::int), '')
+		FROM generate_series(1, 2048) s(x);
+	$$;
+
+	-- Not compressible like above, but small enough to stay in-line.
+	CREATE FUNCTION gen_inline()
+	RETURNS text
+	LANGUAGE sql as $$
+		SELECT string_agg(chr(65 + trunc(25 * random())::int), '')
+		FROM generate_series(1, 1024) s(x);
+	$$;
+
+	-- A varlena short enough to have a one-byte header.
+	CREATE FUNCTION gen_short()
+	RETURNS text
+	LANGUAGE sql as $$
+		SELECT string_agg(chr(65 + trunc(25 * random())::int), '')
+		FROM generate_series(1, 120) s(x);
+	$$;
+
+	CREATE TABLE repack_direct_toast(drop1 int, i int PRIMARY KEY, drop2 int,
+		j text COMPRESSION pglz, k text COMPRESSION pglz) WITH (toast_flavour = 'direct');
+	INSERT INTO repack_direct_toast(drop1, i, drop2, j, k)
+	SELECT 42, gs, 42, gen_external(), gen_compressible(gs) FROM generate_series(1, 10) gs;
+	ALTER TABLE repack_direct_toast DROP COLUMN drop1, DROP COLUMN drop2;
+	ALTER TABLE repack_direct_toast ALTER COLUMN k SET COMPRESSION default;
+	INSERT INTO repack_direct_toast(i, j, k)
+	SELECT gs, gen_external(), gen_compressible(142857) FROM generate_series(11, 20) gs;
+
+	ALTER TABLE repack_direct_toast SET (toast_tuple_target = 128);
+
+	CREATE TABLE relfilenodes(node oid);
+
+	CREATE TABLE data_s1 (i int, j text, j_toast oid, k text, k_toast oid);
+	CREATE TABLE data_s2 (LIKE data_s1);
+}
+
+
+teardown
+{
+	DROP TABLE repack_direct_toast;
+	DROP EXTENSION injection_points;
+	DROP FUNCTION gen_compressible(int);
+	DROP FUNCTION gen_compressible_external(int);
+	DROP FUNCTION gen_external();
+	DROP FUNCTION gen_inline();
+	DROP FUNCTION gen_short();
+
+	DROP TABLE relfilenodes;
+	DROP TABLE data_s1;
+	DROP TABLE data_s2;
+}
+
+session s1
+setup
+{
+	SELECT injection_points_set_local();
+	SELECT injection_points_attach('repack-concurrently-before-lock', 'wait');
+}
+
+# Perform the initial load and wait for s2 to do some data changes.
+step s1_wait_before_lock
+{
+	REPACK (CONCURRENTLY) repack_direct_toast;
+}
+
+# Check the table, after REPACK has completed.  s2 must have saved the data
+# as it was visible to it.  We check that the relfilenode changed in addition
+# to verifying that the actual data matches.
+step s1_check
+{
+	INSERT INTO relfilenodes(node)
+	SELECT c2.relfilenode
+	FROM pg_class c1 JOIN pg_class c2 ON c2.oid = c1.oid OR c2.oid = c1.reltoastrelid
+	WHERE c1.relname='repack_direct_toast';
+
+	SELECT count(DISTINCT node) FROM relfilenodes;
+
+	INSERT INTO data_s1(i, j, j_toast, k, k_toast)
+	SELECT i,
+	j, COALESCE(pg_column_toast_chunk_id(j), 0) AS j_toast,
+	k, COALESCE(pg_column_toast_chunk_id(k), 0) AS k_toast
+	FROM repack_direct_toast;
+
+	-- this should be empty
+	SELECT d1.i, substring(d1.j FOR 12) AS d1_j, substring(d1.k FOR 12) AS d1_k,
+		d2.i, substring(d2.j FOR 12) AS d2_j, substring(d2.k FOR 12) AS d2_k,
+		d1.j_toast as d1_j_tst, d2.j_toast as d2_j_tst,
+		d1.k_toast as d1_k_tst, d2.k_toast AS d2_k_tst
+	FROM data_s1 d1 FULL JOIN data_s2 d2 USING (i, j, k)
+	WHERE d1.i ISNULL OR d2.i ISNULL;
+}
+teardown
+{
+    SELECT injection_points_detach('repack-concurrently-before-lock');
+}
+
+session s2
+
+# Test different kinds of toast data changes.
+step s2_updates
+{
+	DELETE FROM repack_direct_toast WHERE i=1;
+	INSERT INTO repack_direct_toast(i, j, k) VALUES (1, gen_external(), gen_compressible(1));
+
+	-- existing toast data unchanged.  (This covers the case where we
+	-- adjust the toast pointer.)
+	UPDATE repack_direct_toast SET i=i+300 where i % 10 = 2 RETURNING OLD.i, NEW.i;
+
+	-- "j" is here an external indirect, written to the file separately.
+	UPDATE repack_direct_toast SET j=gen_external() where i % 10 = 3 RETURNING OLD.i, NEW.i;
+
+	-- the updated value of "j" is compressed.
+	UPDATE repack_direct_toast SET j=gen_compressible(1), k=k||'' where i % 10 = 4 RETURNING i;
+
+	-- the updated value of "j" is compressed externally.
+	UPDATE repack_direct_toast SET j=gen_compressible_external(2) where i % 10 = 5 RETURNING i;
+
+	-- the updated value of "j" stays inline.
+	UPDATE repack_direct_toast SET j=gen_inline(), k=repeat(k,5) where i % 10 = 6 RETURNING i;
+
+	-- updated value of "j" is a short varlena; "k" is written separately.
+	UPDATE repack_direct_toast SET j=gen_short(), k=gen_external() where i % 10 = 7 RETURNING i;
+}
+
+# Check the table from the perspective of s2.  This saves data so that it can
+# be verified later.
+step s2_check
+{
+	INSERT INTO relfilenodes(node)
+	SELECT c2.relfilenode
+	FROM pg_class c1 JOIN pg_class c2 ON c2.oid = c1.oid OR c2.oid = c1.reltoastrelid
+	WHERE c1.relname='repack_direct_toast';
+
+	INSERT INTO data_s2(i, j, j_toast, k, k_toast)
+	SELECT i, j, COALESCE(pg_column_toast_chunk_id(j), 0) AS j_toast,
+	k, COALESCE(pg_column_toast_chunk_id(k), 0) AS k_toast
+	FROM repack_direct_toast;
+}
+step s2_wakeup_before_lock
+{
+	SELECT injection_points_wakeup('repack-concurrently-before-lock');
+}
+
+# Test if data changes introduced while one session is performing REPACK
+# CONCURRENTLY find their way into the table.
+permutation
+	s1_wait_before_lock
+	s2_updates
+	s2_check
+	s2_wakeup_before_lock
+	s1_check
-- 
2.55.0.970.g62bdec98f9-goog

