From 1b3dfa307bc3e370feef853e75e202c4f0ca99e7 Mon Sep 17 00:00:00 2001
From: Hannu Krosing <hannuk@google.com>
Date: Sun, 23 Aug 2026 20:12:28 +0000
Subject: [PATCH v4 3/9] Implement Direct TOAST core storage reading and
 writing

Implement the physical direct TOAST storage format that eliminates B-Tree
indexes by storing physical TIDs:
- Writing (toast_internals.c): support single-chunk fast-path, flat multi-chunk
  TID arrays, and hierarchical tree DAG construction for large datums (>100 chunks).
- Reading (detoast.c): support direct heap tuple fetching, single-chunk fast path,
  and recursive tree slice retrieval with subtree pruning.
- Deletion (toast_internals.c): implement recursive direct chunk heap deletion.
- Introspection & Helper (toast_compression.c, toast_helper.c): support direct
  pointers in compression detection and tuple size calculation.
- Add comprehensive regression test suite (direct_toast.sql / direct_toast.out).
---
 src/backend/access/common/detoast.c           | 385 ++++++++++--
 src/backend/access/common/toast_compression.c |   9 +
 src/backend/access/common/toast_internals.c   | 414 ++++++++++++-
 src/backend/access/table/toast_helper.c       |   6 +-
 src/backend/utils/adt/arrayfuncs.c            |   6 +
 src/backend/utils/misc/guc_parameters.dat     |   7 +
 src/include/access/detoast.h                  |  11 +
 src/test/regress/expected/cluster.out         |   1 +
 src/test/regress/expected/direct_toast.out    | 565 ++++++++++++++++++
 src/test/regress/expected/psql.out            |  14 +-
 src/test/regress/parallel_schedule            |   2 +-
 src/test/regress/sql/direct_toast.sql         | 397 ++++++++++++
 12 files changed, 1768 insertions(+), 49 deletions(-)
 create mode 100644 src/test/regress/expected/direct_toast.out
 create mode 100644 src/test/regress/sql/direct_toast.sql

diff --git a/src/backend/access/common/detoast.c b/src/backend/access/common/detoast.c
index 305d3cfe0fc..b733a794cd9 100644
--- a/src/backend/access/common/detoast.c
+++ b/src/backend/access/common/detoast.c
@@ -14,13 +14,18 @@
 #include "postgres.h"
 
 #include "access/detoast.h"
+#include "access/heapam.h"
+#include "access/heaptoast.h"
 #include "access/table.h"
 #include "access/tableam.h"
 #include "access/toast_internals.h"
+#include "catalog/pg_type.h"
 #include "common/int.h"
 #include "common/pg_lzcompress.h"
+#include "storage/bufmgr.h"
 #include "utils/expandeddatum.h"
 #include "utils/rel.h"
+#include "utils/array.h"
 
 static varlena *toast_fetch_datum_slice(varlena *attr,
 										int32 sliceoffset,
@@ -37,6 +42,56 @@ toast_decompress_datum(varlena *attr)
 {
 	return toast_decompress_datum_slice(attr, -1);
 }
+static void toast_fetch_datum_direct_slice_recursive(Relation toastrel, ItemPointer tid,
+													 varlena *result, int32 *logical_offset,
+													 int32 sliceoffset, int32 slicelength,
+													 TupleTableSlot *slot);
+
+/*
+ * Unpacked metadata from either plain (varatt_external) or direct (varatt_direct)
+ * on-disk TOAST pointers.
+ */
+typedef struct ToastExternalMetadata
+{
+	int32		extsize;
+	uint32		compress_method;
+	bool		is_compressed;
+	Oid			toastrelid;
+	bool		is_direct;
+	Oid8		valueid;
+	struct varatt_direct direct_tp;
+} ToastExternalMetadata;
+
+static inline void
+toast_get_external_metadata(varlena *attr, ToastExternalMetadata *meta)
+{
+	if (VARATT_IS_EXTERNAL_DIRECT(attr))
+	{
+		VARATT_EXTERNAL_GET_POINTER_DIRECT(meta->direct_tp, attr);
+		meta->extsize = VARATT_DIRECT_GET_EXTSIZE(meta->direct_tp);
+		meta->compress_method = VARATT_DIRECT_GET_COMPRESS_METHOD(meta->direct_tp);
+		meta->is_compressed = VARATT_DIRECT_IS_COMPRESSED(meta->direct_tp);
+		meta->toastrelid = meta->direct_tp.va_toastrelid;
+		meta->is_direct = true;
+		meta->valueid = 0;
+	}
+	else if (VARATT_IS_EXTERNAL_ONDISK(attr))
+	{
+		toast_external_data toast_ext_data;
+
+		toast_external_info_get(attr, &toast_ext_data);
+		meta->extsize = VARATT_EXTINFO_GET_EXTSIZE(toast_ext_data.extinfo);
+		meta->compress_method = VARATT_EXTINFO_GET_COMPRESS_METHOD(toast_ext_data.extinfo);
+		meta->is_compressed = VARATT_EXTINFO_IS_COMPRESSED(toast_ext_data.extinfo, toast_ext_data.rawsize);
+		meta->toastrelid = toast_ext_data.toastrelid;
+		meta->is_direct = false;
+		meta->valueid = toast_ext_data.valueid;
+	}
+	else
+	{
+		elog(ERROR, "toast_get_external_metadata called for unsupported datum");
+	}
+}
 
 /* ----------
  * detoast_external_attr -
@@ -55,10 +110,10 @@ detoast_external_attr(varlena *attr)
 {
 	varlena    *result;
 
-	if (VARATT_IS_EXTERNAL_ONDISK(attr))
+	if (VARATT_IS_EXTERNAL_ONDISK(attr) || VARATT_IS_EXTERNAL_DIRECT(attr))
 	{
 		/*
-		 * This is an external stored plain value
+		 * This is an external stored plain or direct value
 		 */
 		result = toast_fetch_datum(attr);
 	}
@@ -160,21 +215,15 @@ detoast_attr_slice(varlena *attr,
 	else if (pg_add_s32_overflow(sliceoffset, slicelength, &slicelimit))
 		slicelength = slicelimit = -1;
 
-	if (VARATT_IS_EXTERNAL_ONDISK(attr))
+	if (VARATT_IS_EXTERNAL_ONDISK(attr) || VARATT_IS_EXTERNAL_DIRECT(attr))
 	{
-		toast_external_data toast_ext_data;
-		int32		extsize;
-		uint32		compress_method;
-		bool		is_compressed;
+		ToastExternalMetadata meta;
 		int32		max_size = -1;
 
-		toast_external_info_get(attr, &toast_ext_data);
-		extsize = VARATT_EXTINFO_GET_EXTSIZE(toast_ext_data.extinfo);
-		compress_method = VARATT_EXTINFO_GET_COMPRESS_METHOD(toast_ext_data.extinfo);
-		is_compressed = VARATT_EXTINFO_IS_COMPRESSED(toast_ext_data.extinfo, toast_ext_data.rawsize);
+		toast_get_external_metadata(attr, &meta);
 
 		/* fast path for non-compressed external datums */
-		if (!is_compressed)
+		if (!meta.is_compressed)
 			return toast_fetch_datum_slice(attr, sliceoffset, slicelength);
 
 		/*
@@ -184,7 +233,7 @@ detoast_attr_slice(varlena *attr,
 		 */
 		if (slicelimit >= 0)
 		{
-			max_size = extsize;
+			max_size = meta.extsize;
 
 			/*
 			 * Determine maximum amount of compressed data needed for a prefix
@@ -195,7 +244,7 @@ detoast_attr_slice(varlena *attr,
 			 * determine how much compressed data we need to be sure of being
 			 * able to decompress the required slice.
 			 */
-			if (compress_method == TOAST_PGLZ_COMPRESSION_ID)
+			if (meta.compress_method == TOAST_PGLZ_COMPRESSION_ID)
 				max_size = pglz_maximum_compressed_size(slicelimit, max_size);
 		}
 
@@ -295,13 +344,11 @@ detoast_attr_slice(varlena *attr,
 	return result;
 }
 
-
-
 /* ----------
  * toast_fetch_datum_slice -
  *
  *	Reconstruct a segment of a Datum from the chunks saved
- *	in the toast relation
+ *	in the toast relation (supports both plain index-based and direct TOAST).
  *
  *	Note that this function supports non-compressed external datums
  *	and compressed external datums (in which case the requested slice
@@ -314,27 +361,19 @@ toast_fetch_datum_slice(varlena *attr, int32 sliceoffset,
 {
 	Relation	toastrel;
 	varlena    *result;
-	toast_external_data toast_ext_data;
 	int32		attrsize;
-	Oid			toastrelid;
-	Oid8		valueid;
-	bool		is_compressed;
-
-	if (!VARATT_IS_EXTERNAL_ONDISK(attr))
-		elog(ERROR, "toast_fetch_datum_slice shouldn't be called for non-ondisk datums");
+	ToastExternalMetadata meta;
 
-	toast_external_info_get(attr, &toast_ext_data);
-	attrsize = VARATT_EXTINFO_GET_EXTSIZE(toast_ext_data.extinfo);
-	toastrelid = toast_ext_data.toastrelid;
-	valueid = toast_ext_data.valueid;
-	is_compressed = VARATT_EXTINFO_IS_COMPRESSED(toast_ext_data.extinfo, toast_ext_data.rawsize);
+	toast_get_external_metadata(attr, &meta);
 
 	/*
 	 * It's nonsense to fetch slices of a compressed datum unless when it's a
 	 * prefix -- this isn't lo_* we can't return a compressed datum which is
 	 * meaningful to toast later.
 	 */
-	Assert(!is_compressed || 0 == sliceoffset);
+	Assert(!meta.is_compressed || 0 == sliceoffset);
+
+	attrsize = meta.extsize;
 
 	if (sliceoffset >= attrsize)
 	{
@@ -347,7 +386,7 @@ toast_fetch_datum_slice(varlena *attr, int32 sliceoffset,
 	 * space required by va_tcinfo, which is stored at the beginning as an
 	 * int32 value.
 	 */
-	if (is_compressed && slicelength > 0)
+	if (meta.is_compressed && slicelength > 0)
 		slicelength = slicelength + sizeof(int32);
 
 	/*
@@ -360,7 +399,7 @@ toast_fetch_datum_slice(varlena *attr, int32 sliceoffset,
 
 	result = (varlena *) palloc(slicelength + VARHDRSZ);
 
-	if (is_compressed)
+	if (meta.is_compressed)
 		SET_VARSIZE_COMPRESSED(result, slicelength + VARHDRSZ);
 	else
 		SET_VARSIZE(result, slicelength + VARHDRSZ);
@@ -369,12 +408,75 @@ toast_fetch_datum_slice(varlena *attr, int32 sliceoffset,
 		return result;			/* Can save a lot of work at this point! */
 
 	/* Open the toast relation */
-	toastrel = table_open(toastrelid, AccessShareLock);
+	toastrel = table_open(meta.toastrelid, AccessShareLock);
 
-	/* Fetch all chunks */
-	table_relation_fetch_toast_slice(toastrel, valueid,
-									 attrsize, sliceoffset, slicelength,
-									 result);
+	if (meta.is_direct)
+	{
+		/*
+		 * Fast path for single-chunk direct TOAST: fetch directly via
+		 * heap_fetch without allocating/dropping a TupleTableSlot.
+		 */
+		if (attrsize <= TOAST_MAX_CHUNK_SIZE(TupleDescAttr(toastrel->rd_att, 0)->atttypid))
+		{
+			HeapTupleData tup;
+			Buffer		buffer = InvalidBuffer;
+			bool		isnull;
+			Pointer		chunk;
+			int32		chunk_size;
+			char	   *chunk_data;
+
+			tup.t_self = meta.direct_tp.va_tid;
+			if (!heap_fetch(toastrel, get_toast_snapshot(), &tup, &buffer, false))
+				elog(ERROR, "failed to fetch toast tuple by TID");
+
+			chunk = DatumGetPointer(fastgetattr(&tup, 3, toastrel->rd_att, &isnull));
+			if (isnull)
+				elog(ERROR, "unexpected NULL chunk_data in direct toast chunk");
+
+			if (!VARATT_IS_EXTENDED(chunk))
+			{
+				chunk_size = VARSIZE(chunk) - VARHDRSZ;
+				chunk_data = VARDATA(chunk);
+			}
+			else if (VARATT_IS_SHORT(chunk))
+			{
+				chunk_size = VARSIZE_SHORT(chunk) - VARHDRSZ_SHORT;
+				chunk_data = VARDATA_SHORT(chunk);
+			}
+			else
+				elog(ERROR, "unexpected type of toast chunk");
+
+			if (sliceoffset >= chunk_size)
+			{
+				slicelength = 0;
+				sliceoffset = 0;
+			}
+			else if (sliceoffset + slicelength > chunk_size || slicelength < 0)
+				slicelength = chunk_size - sliceoffset;
+
+			if (slicelength > 0)
+				memcpy(VARDATA(result), chunk_data + sliceoffset, slicelength);
+
+			ReleaseBuffer(buffer);
+		}
+		else
+		{
+			TupleTableSlot *slot = table_slot_create(toastrel, NULL);
+			int32		logical_offset = 0;
+
+			toast_fetch_datum_direct_slice_recursive(toastrel, &meta.direct_tp.va_tid,
+													 result, &logical_offset,
+													 sliceoffset, slicelength, slot);
+			ExecDropSingleTupleTableSlot(slot);
+		}
+	}
+	else
+	{
+		/* Fetch all chunks via Table AM index scan */
+		table_relation_fetch_toast_slice(toastrel, meta.valueid,
+										 attrsize, sliceoffset, slicelength,
+										 result);
+	}
 
 	/* Close toast table */
 	table_close(toastrel, AccessShareLock);
@@ -434,6 +536,13 @@ toast_raw_datum_size(Datum value)
 		toast_external_info_get(attr, &toast_ext_data);
 		result = toast_ext_data.rawsize;
 	}
+	else if (VARATT_IS_EXTERNAL_DIRECT(attr))
+	{
+		struct varatt_direct toast_pointer;
+
+		VARATT_EXTERNAL_GET_POINTER_DIRECT(toast_pointer, attr);
+		result = toast_pointer.va_rawsize;
+	}
 	else if (VARATT_IS_EXTERNAL_INDIRECT(attr))
 	{
 		varatt_indirect toast_pointer;
@@ -494,6 +603,13 @@ toast_datum_size(Datum value)
 		toast_external_info_get(attr, &toast_ext_data);
 		result = VARATT_EXTINFO_GET_EXTSIZE(toast_ext_data.extinfo);
 	}
+	else if (VARATT_IS_EXTERNAL_DIRECT(attr))
+	{
+		struct varatt_direct toast_pointer;
+
+		VARATT_EXTERNAL_GET_POINTER_DIRECT(toast_pointer, attr);
+		result = VARATT_DIRECT_GET_EXTSIZE(toast_pointer);
+	}
 	else if (VARATT_IS_EXTERNAL_INDIRECT(attr))
 	{
 		varatt_indirect toast_pointer;
@@ -523,3 +639,198 @@ toast_datum_size(Datum value)
 	}
 	return result;
 }
+
+/*
+ * Helper to copy overlapping chunk slice into detoasted result varlena.
+ */
+static inline void
+toast_slice_copy_chunk(struct varlena *result, const char *chunk_data,
+					   int32 chunk_size, int32 *logical_offset,
+					   int32 req_start, int32 req_end)
+{
+	int32		chunk_start = *logical_offset;
+	int32		chunk_end = chunk_start + chunk_size;
+
+	*logical_offset = chunk_end;
+
+	if (chunk_end > req_start && chunk_start < req_end)
+	{
+		int32		copy_start = Max(chunk_start, req_start);
+		int32		copy_end = Min(chunk_end, req_end);
+		int32		copy_len = copy_end - copy_start;
+
+		if (copy_len > 0)
+		{
+			int32		src_offset = copy_start - chunk_start;
+			int32		dest_offset = copy_start - req_start;
+
+			memcpy(VARDATA(result) + dest_offset, chunk_data + src_offset, copy_len);
+		}
+	}
+}
+
+/*
+ * Recursively traverse and fetch slices from a direct TOAST tree/DAG.
+ */
+static void
+toast_fetch_datum_direct_slice_recursive(Relation toastrel, ItemPointer tid,
+										 struct varlena *result, int32 *logical_offset,
+										 int32 sliceoffset, int32 slicelength,
+										 TupleTableSlot *slot)
+{
+	Snapshot	snapshot = get_toast_snapshot();
+	Datum		data_datum;
+	Datum		tids_datum;
+	Datum		offsets_datum;
+	bool		is_null_data;
+	bool		is_null_tids;
+	bool		is_null_offsets = true;
+
+	if (!table_tuple_fetch_row_version(toastrel, tid, snapshot, slot))
+	{
+		elog(ERROR, "failed to fetch toast tuple by TID");
+	}
+
+	data_datum = slot_getattr(slot, 3, &is_null_data);
+	tids_datum = slot_getattr(slot, 4, &is_null_tids);
+	if (is_null_tids)
+	{
+		/* Leaf chunk: copy data slice */
+		if (!is_null_data)
+		{
+			struct varlena *data_val = PG_DETOAST_DATUM(data_datum);
+			int32		chunk_size = VARSIZE_ANY_EXHDR(data_val);
+			int32		req_start = sliceoffset;
+			int32		req_end = sliceoffset + slicelength;
+
+			toast_slice_copy_chunk(result, VARDATA_ANY(data_val), chunk_size,
+								   logical_offset, req_start, req_end);
+		}
+		ExecClearTuple(slot);
+	}
+	else
+	{
+		ArrayType  *arr = DatumGetArrayTypePCopy(tids_datum);
+		Datum	   *elems;
+		bool	   *nulls;
+		int			nelems;
+		int			i;
+
+		if (slot->tts_tupleDescriptor->natts >= 5)
+			offsets_datum = slot_getattr(slot, 5, &is_null_offsets);
+
+		deconstruct_array_builtin(arr, TIDOID, &elems, &nulls, &nelems);
+
+		if (!is_null_offsets)
+		{
+			/*
+			 * Tree-structured interior node with chunk_tid_offsets.
+			 * Use offsets to prune subtrees that don't overlap the requested slice.
+			 */
+			ArrayType  *arr_offsets = DatumGetArrayTypePCopy(offsets_datum);
+			Datum	   *offset_elems;
+			bool	   *offset_nulls;
+			int			noffsets;
+			int64		req_start = sliceoffset;
+			int64		req_end = (slicelength < 0) ? PG_INT64_MAX : ((int64) sliceoffset + slicelength);
+
+			deconstruct_array_builtin(arr_offsets, INT8OID, &offset_elems, &offset_nulls, &noffsets);
+			Assert(noffsets == nelems + 1);
+
+			ExecClearTuple(slot);
+
+			for (i = 0; i < nelems; i++)
+			{
+				int64		child_start = DatumGetInt64(offset_elems[i]);
+				int64		child_end = DatumGetInt64(offset_elems[i + 1]);
+
+				if (child_end <= req_start || child_start >= req_end)
+				{
+					/* Subtree does not intersect slice range; skip it */
+					*logical_offset = (int32) child_end;
+					continue;
+				}
+
+				*logical_offset = (int32) child_start;
+				toast_fetch_datum_direct_slice_recursive(toastrel,
+														 (ItemPointer) DatumGetPointer(elems[i]),
+														 result, logical_offset,
+														 sliceoffset, slicelength,
+														 slot);
+				*logical_offset = (int32) child_end;
+			}
+
+			pfree(offset_elems);
+			pfree(offset_nulls);
+			pfree(arr_offsets);
+		}
+		else
+		{
+			/*
+			 * Flat direct TOAST: chunks 0 to nelems-1 are leaf data chunks
+			 * of size TOAST_MAX_CHUNK_SIZE, and the current chunk contains the
+			 * final chunk_data (chunk nelems).
+			 */
+			int32		max_chunk_size = TOAST_MAX_CHUNK_SIZE(TupleDescAttr(toastrel->rd_att, 0)->atttypid);
+			int32		chunk_size = 0;
+			char	   *chunk_data = NULL;
+			struct varlena *data_val = NULL;
+			int64		req_start = sliceoffset;
+			int64		req_end = (slicelength < 0) ? PG_INT64_MAX : ((int64) sliceoffset + slicelength);
+
+			if (!is_null_data)
+			{
+				data_val = PG_DETOAST_DATUM_COPY(data_datum);
+				chunk_size = VARSIZE_ANY_EXHDR(data_val);
+				chunk_data = VARDATA_ANY(data_val);
+			}
+
+			ExecClearTuple(slot);
+
+			for (i = 0; i < nelems; i++)
+			{
+				int64		child_start = (int64) i * max_chunk_size;
+				int64		child_end = child_start + max_chunk_size;
+
+				if (child_end <= req_start || child_start >= req_end)
+				{
+					/* Chunk does not intersect requested slice */
+					*logical_offset = (int32) child_end;
+					continue;
+				}
+
+				*logical_offset = (int32) child_start;
+				toast_fetch_datum_direct_slice_recursive(toastrel,
+														 (ItemPointer) DatumGetPointer(elems[i]),
+														 result, logical_offset,
+														 sliceoffset, slicelength,
+														 slot);
+				*logical_offset = (int32) child_end;
+			}
+
+			if (chunk_size > 0)
+			{
+				int64		root_start = (int64) nelems * max_chunk_size;
+				int64		root_end = root_start + chunk_size;
+
+				if (root_end > req_start && root_start < req_end)
+				{
+					int32		copy_req_end = (slicelength < 0) ? PG_INT32_MAX : (sliceoffset + slicelength);
+
+					*logical_offset = (int32) root_start;
+					toast_slice_copy_chunk(result, chunk_data, chunk_size,
+										   logical_offset, sliceoffset, copy_req_end);
+				}
+				else
+					*logical_offset = (int32) root_end;
+			}
+
+			if (data_val)
+				pfree(data_val);
+		}
+
+		pfree(elems);
+		pfree(nulls);
+		pfree(arr);
+	}
+}
diff --git a/src/backend/access/common/toast_compression.c b/src/backend/access/common/toast_compression.c
index e4ba46cc9f4..70bcd1a8a88 100644
--- a/src/backend/access/common/toast_compression.c
+++ b/src/backend/access/common/toast_compression.c
@@ -259,6 +259,15 @@ toast_get_compression_id(varlena *attr)
 		if (VARATT_EXTINFO_IS_COMPRESSED(toast_ext_data.extinfo, toast_ext_data.rawsize))
 			cmid = VARATT_EXTINFO_GET_COMPRESS_METHOD(toast_ext_data.extinfo);
 	}
+	else if (VARATT_IS_EXTERNAL_DIRECT(attr))
+	{
+		struct varatt_direct toast_pointer;
+
+		VARATT_EXTERNAL_GET_POINTER_DIRECT(toast_pointer, attr);
+
+		if (VARATT_DIRECT_IS_COMPRESSED(toast_pointer))
+			cmid = VARATT_DIRECT_GET_COMPRESS_METHOD(toast_pointer);
+	}
 	else if (VARATT_IS_COMPRESSED(attr))
 		cmid = VARDATA_COMPRESSED_GET_COMPRESS_METHOD(attr);
 
diff --git a/src/backend/access/common/toast_internals.c b/src/backend/access/common/toast_internals.c
index a4e8a096254..ef16ca1d762 100644
--- a/src/backend/access/common/toast_internals.c
+++ b/src/backend/access/common/toast_internals.c
@@ -21,13 +21,31 @@
 #include "access/toast_internals.h"
 #include "access/xact.h"
 #include "catalog/catalog.h"
+#include "catalog/pg_type.h"
 #include "miscadmin.h"
 #include "utils/fmgroids.h"
 #include "utils/rel.h"
 #include "utils/snapmgr.h"
+#include "utils/array.h"
+
+int			toast_flavour = TOAST_FLAVOUR_PLAIN;
+
+#define DIRECT_TOAST_TREE_THRESHOLD	100
+#define DIRECT_TOAST_FANOUT			50
+
+typedef struct DirectToastItem
+{
+	ItemPointerData tid;
+	int64		start_offset;
+	int64		end_offset;
+} DirectToastItem;
 
 static bool toastrel_valueid_exists(Relation toastrel, Oid8 valueid);
 static bool toastid_valueid_exists(Oid toastrelid, Oid8 valueid);
+static Datum toast_save_datum_direct(Relation rel, Datum value,
+									 varlena *oldexternal, int options);
+static void toast_delete_datum_direct(Relation rel, Datum value, bool is_speculative);
+static void toast_delete_datum_direct_recursive(Relation toastrel, ItemPointer tid, bool is_speculative);
 
 /* ----------
  * toast_compress_datum -
@@ -191,6 +209,9 @@ toast_save_datum(Relation rel, Datum value,
 
 	Assert(!VARATT_IS_EXTERNAL(dval));
 
+	if (RelationGetToastFlavour(rel) == TOAST_FLAVOUR_DIRECT)
+		return toast_save_datum_direct(rel, value, oldexternal, options);
+
 	/*
 	 * Open the toast relation and its indexes.  We can use the index to check
 	 * uniqueness of the OID we assign to the toasted item, even though it has
@@ -300,8 +321,8 @@ toast_save_datum(Relation rel, Datum value,
 	while (data_todo > 0)
 	{
 		HeapTuple	toasttup;
-		Datum		t_values[3];
-		bool		t_isnull[3] = {0};
+		Datum		t_values[5];
+		bool		t_isnull[5] = {0};
 		union
 		{
 			alignas(int32) varlena hdr;
@@ -330,6 +351,10 @@ toast_save_datum(Relation rel, Datum value,
 		SET_VARSIZE(&chunk_data, chunk_size + VARHDRSZ);
 		memcpy(VARDATA(&chunk_data), data_p, chunk_size);
 		t_values[2] = PointerGetDatum(&chunk_data);
+		t_values[3] = PointerGetDatum(NULL);
+		t_isnull[3] = true;
+		t_values[4] = PointerGetDatum(NULL);
+		t_isnull[4] = true;
 
 		toasttup = heap_form_tuple(toasttupDesc, t_values, t_isnull);
 
@@ -450,6 +475,12 @@ toast_delete_datum(Relation rel, Datum value, bool is_speculative)
 	int			num_indexes;
 	int			validIndex;
 
+	if (VARATT_IS_EXTERNAL_DIRECT(attr))
+	{
+		toast_delete_datum_direct(rel, value, is_speculative);
+		return;
+	}
+
 	if (!VARATT_IS_EXTERNAL_ONDISK(attr))
 		return;
 
@@ -713,3 +744,382 @@ get_toast_snapshot(void)
 
 	return &SnapshotToastData;
 }
+
+/*
+ * Context struct for direct TOAST write operations.
+ */
+typedef struct DirectToastWriteState
+{
+	Relation	toastrel;
+	TupleDesc	toasttupDesc;
+	CommandId	mycid;
+	int			options;
+	int32		chunk_seq;
+	int32		max_chunk_size;
+} DirectToastWriteState;
+
+/*
+ * Helper to form and insert a direct TOAST chunk tuple into the toast table.
+ * Encapsulates array construction and memory management for child TIDs and offsets.
+ */
+static inline ItemPointerData
+toast_direct_insert_chunk(DirectToastWriteState *state,
+						  const char *chunk_data_p, int32 chunk_size,
+						  const ItemPointerData *tids, int num_tids,
+						  const int64 *offsets, int num_offsets)
+{
+	Datum		t_values[5];
+	bool		t_isnull[5];
+	HeapTuple	toasttup;
+	ItemPointerData tid;
+	ArrayType  *tid_array = NULL;
+	ArrayType  *offset_array = NULL;
+	union
+	{
+		struct varlena hdr;
+		char		data[TOAST_OID_MAX_CHUNK_SIZE + VARHDRSZ];
+		int32		align_it;
+	}			chunk_buf;
+
+	t_values[0] = (Datum) 0;
+	t_isnull[0] = true;			/* chunk_id is NULL */
+
+	t_values[1] = Int32GetDatum(state->chunk_seq++);
+	t_isnull[1] = false;
+
+	if (chunk_data_p && chunk_size > 0)
+	{
+		SET_VARSIZE(&chunk_buf, chunk_size + VARHDRSZ);
+		memcpy(VARDATA(&chunk_buf), chunk_data_p, chunk_size);
+		t_values[2] = PointerGetDatum(&chunk_buf);
+		t_isnull[2] = false;
+	}
+	else
+	{
+		t_values[2] = (Datum) 0;
+		t_isnull[2] = true;
+	}
+
+	if (tids && num_tids > 0)
+	{
+		Datum	   *tids_datums = palloc(sizeof(Datum) * num_tids);
+
+		for (int i = 0; i < num_tids; i++)
+			tids_datums[i] = PointerGetDatum(&tids[i]);
+		tid_array = construct_array_builtin(tids_datums, num_tids, TIDOID);
+		pfree(tids_datums);
+
+		t_values[3] = PointerGetDatum(tid_array);
+		t_isnull[3] = false;
+	}
+	else
+	{
+		t_values[3] = (Datum) 0;
+		t_isnull[3] = true;
+	}
+
+	if (offsets && num_offsets > 0)
+	{
+		Datum	   *offsets_datums = palloc(sizeof(Datum) * num_offsets);
+
+		for (int i = 0; i < num_offsets; i++)
+			offsets_datums[i] = Int64GetDatum(offsets[i]);
+		offset_array = construct_array_builtin(offsets_datums, num_offsets, INT8OID);
+		pfree(offsets_datums);
+
+		t_values[4] = PointerGetDatum(offset_array);
+		t_isnull[4] = false;
+	}
+	else
+	{
+		t_values[4] = (Datum) 0;
+		t_isnull[4] = true;
+	}
+
+	toasttup = heap_form_tuple(state->toasttupDesc, t_values, t_isnull);
+	heap_insert(state->toastrel, toasttup, state->mycid, state->options, NULL);
+	tid = toasttup->t_self;
+	heap_freetuple(toasttup);
+
+	if (tid_array)
+		pfree(tid_array);
+	if (offset_array)
+		pfree(offset_array);
+
+	return tid;
+}
+
+/*
+ * Single-chunk direct TOAST write path (Tier 1: <= ~2 kB).
+ * Returns the physical TID of the single chunk.
+ */
+static ItemPointerData
+toast_save_direct_single(DirectToastWriteState *state,
+						 const char *data_p, int32 data_len)
+{
+	return toast_direct_insert_chunk(state, data_p, data_len,
+									 NULL, 0, NULL, 0);
+}
+
+/*
+ * Flat multi-chunk direct TOAST write path (Tier 2: <= 100 chunks, up to ~200 kB).
+ * Writes leaf chunks 0 to N-2, and writes chunk N-1 containing the remaining data
+ * plus an array of TIDs of chunks 0 to N-2. Returns chunk N-1 TID.
+ */
+static ItemPointerData
+toast_save_direct_flat(DirectToastWriteState *state,
+					   const char *data_p, int32 data_todo, int total_chunks)
+{
+	ItemPointerData *tids = palloc(sizeof(ItemPointerData) * (total_chunks - 1));
+	ItemPointerData root_tid;
+	int32		chunk_size;
+
+	/* Insert leaf chunks 0 to N-2 */
+	for (int i = 0; i < total_chunks - 1; i++)
+	{
+		CHECK_FOR_INTERRUPTS();
+		chunk_size = Min(state->max_chunk_size, data_todo);
+
+		tids[i] = toast_direct_insert_chunk(state, data_p, chunk_size,
+											NULL, 0, NULL, 0);
+		data_todo -= chunk_size;
+		data_p += chunk_size;
+	}
+
+	/* Insert final chunk containing remaining data and previous TIDs */
+	CHECK_FOR_INTERRUPTS();
+	chunk_size = data_todo;
+	Assert(chunk_size <= state->max_chunk_size);
+
+	root_tid = toast_direct_insert_chunk(state, data_p, chunk_size,
+										 tids, total_chunks - 1,
+										 NULL, 0);
+	pfree(tids);
+	return root_tid;
+}
+
+/*
+ * Hierarchical tree DAG direct TOAST write path (Tier 3: > 100 chunks, up to 1 GB+).
+ * Writes all leaf data chunks first, then recursively builds interior nodes
+ * recording child TIDs and byte offset boundaries. Returns the top root TID.
+ */
+static ItemPointerData
+toast_save_direct_tree(DirectToastWriteState *state,
+					   const char *data_p, int32 data_todo, int total_chunks)
+{
+	DirectToastItem *items = palloc(sizeof(DirectToastItem) * total_chunks);
+	int			num_items = total_chunks;
+	int64		cur_offset = 0;
+	ItemPointerData root_tid;
+
+	/* Step 1: Write all leaf data chunks */
+	for (int i = 0; i < total_chunks; i++)
+	{
+		int32		chunk_size;
+
+		CHECK_FOR_INTERRUPTS();
+		chunk_size = Min(state->max_chunk_size, data_todo);
+
+		items[i].tid = toast_direct_insert_chunk(state, data_p, chunk_size,
+												 NULL, 0, NULL, 0);
+		items[i].start_offset = cur_offset;
+		cur_offset += chunk_size;
+		items[i].end_offset = cur_offset;
+
+		data_todo -= chunk_size;
+		data_p += chunk_size;
+	}
+
+	/* Step 2: Build tree levels until only 1 root node remains */
+	while (num_items > 1)
+	{
+		int			num_parents = (num_items + DIRECT_TOAST_FANOUT - 1) / DIRECT_TOAST_FANOUT;
+		DirectToastItem *parent_items = palloc(sizeof(DirectToastItem) * num_parents);
+
+		for (int p = 0; p < num_parents; p++)
+		{
+			int			start_idx = p * DIRECT_TOAST_FANOUT;
+			int			count = Min(DIRECT_TOAST_FANOUT, num_items - start_idx);
+			ItemPointerData child_tids[DIRECT_TOAST_FANOUT];
+			int64		child_offsets[DIRECT_TOAST_FANOUT + 1];
+
+			CHECK_FOR_INTERRUPTS();
+
+			for (int k = 0; k < count; k++)
+			{
+				child_tids[k] = items[start_idx + k].tid;
+				child_offsets[k] = items[start_idx + k].start_offset;
+			}
+			child_offsets[count] = items[start_idx + count - 1].end_offset;
+
+			parent_items[p].tid = toast_direct_insert_chunk(state, NULL, 0,
+															child_tids, count,
+															child_offsets, count + 1);
+			parent_items[p].start_offset = items[start_idx].start_offset;
+			parent_items[p].end_offset = items[start_idx + count - 1].end_offset;
+		}
+
+		pfree(items);
+		items = parent_items;
+		num_items = num_parents;
+	}
+
+	root_tid = items[0].tid;
+	pfree(items);
+	return root_tid;
+}
+
+/*
+ * Direct TOAST write path.
+ * Stores the TID of the root chunk directly in the varlena header.
+ * Dispatches to single-chunk, flat multi-chunk, or tree-structured writers.
+ */
+static Datum
+toast_save_datum_direct(Relation rel, Datum value,
+						struct varlena *oldexternal, int options)
+{
+	DirectToastWriteState state;
+	struct varatt_direct toast_pointer;
+	struct varlena *result;
+	char	   *data_p;
+	int32		data_todo;
+	Pointer		dval = DatumGetPointer(value);
+	int			total_chunks;
+
+	Assert(!VARATT_IS_EXTERNAL(dval));
+	memset(&toast_pointer, 0, sizeof(toast_pointer));
+
+	state.toastrel = table_open(rel->rd_rel->reltoastrelid, RowExclusiveLock);
+	state.toasttupDesc = state.toastrel->rd_att;
+	state.mycid = GetCurrentCommandId(true);
+	state.options = options;
+	state.chunk_seq = 0;
+	state.max_chunk_size = TOAST_MAX_CHUNK_SIZE(TupleDescAttr(state.toasttupDesc, 0)->atttypid);
+
+	if (VARATT_IS_SHORT(dval))
+	{
+		data_p = VARDATA_SHORT(dval);
+		data_todo = VARSIZE_SHORT(dval) - VARHDRSZ_SHORT;
+		toast_pointer.va_rawsize = data_todo + VARHDRSZ;
+		toast_pointer.va_extinfo = data_todo;
+	}
+	else if (VARATT_IS_COMPRESSED(dval))
+	{
+		uint32		cmid = VARDATA_COMPRESSED_GET_COMPRESS_METHOD(dval);
+
+		data_p = VARDATA(dval);
+		data_todo = VARSIZE(dval) - VARHDRSZ;
+		toast_pointer.va_rawsize = VARDATA_COMPRESSED_GET_EXTSIZE(dval) + VARHDRSZ;
+		toast_pointer.va_extinfo = data_todo | (cmid << VARLENA_EXTSIZE_BITS);
+	}
+	else
+	{
+		data_p = VARDATA(dval);
+		data_todo = VARSIZE(dval) - VARHDRSZ;
+		toast_pointer.va_rawsize = VARSIZE(dval);
+		toast_pointer.va_extinfo = data_todo;
+	}
+
+	if (OidIsValid(rel->rd_toastoid))
+		toast_pointer.va_toastrelid = rel->rd_toastoid;
+	else
+		toast_pointer.va_toastrelid = RelationGetRelid(state.toastrel);
+
+	total_chunks = ((data_todo - 1) / state.max_chunk_size) + 1;
+
+	if (total_chunks == 1)
+		toast_pointer.va_tid = toast_save_direct_single(&state, data_p, data_todo);
+	else if (total_chunks <= DIRECT_TOAST_TREE_THRESHOLD)
+		toast_pointer.va_tid = toast_save_direct_flat(&state, data_p, data_todo, total_chunks);
+	else
+		toast_pointer.va_tid = toast_save_direct_tree(&state, data_p, data_todo, total_chunks);
+
+	table_close(state.toastrel, NoLock);
+
+	result = (struct varlena *) palloc(DIRECT_POINTER_SIZE);
+	SET_VARTAG_EXTERNAL(result, VARTAG_DIRECT);
+	memcpy(VARDATA_EXTERNAL(result), &toast_pointer, sizeof(toast_pointer));
+
+	Assert(VARATT_IS_EXTERNAL(result));
+	Assert(VARTAG_EXTERNAL(result) == VARTAG_DIRECT);
+	Assert(VARSIZE_EXTERNAL(result) == DIRECT_POINTER_SIZE);
+
+	return PointerGetDatum(result);
+}
+
+/*
+ * Direct TOAST delete path.
+ */
+static void
+toast_delete_datum_direct(Relation rel, Datum value, bool is_speculative)
+{
+	struct varlena *attr = (varlena *) DatumGetPointer(value);
+	struct varatt_direct toast_pointer;
+	Relation	toastrel;
+
+	VARATT_EXTERNAL_GET_POINTER_DIRECT(toast_pointer, attr);
+
+	toastrel = table_open(toast_pointer.va_toastrelid, RowExclusiveLock);
+
+	toast_delete_datum_direct_recursive(toastrel, &toast_pointer.va_tid, is_speculative);
+
+	table_close(toastrel, RowExclusiveLock);
+}
+
+/*
+ * Recursively delete direct TOAST tuples.
+ */
+static void
+toast_delete_datum_direct_recursive(Relation toastrel, ItemPointer tid, bool is_speculative)
+{
+	TupleTableSlot *slot;
+	Snapshot	snapshot = get_toast_snapshot();
+	Datum		tids_datum;
+	bool		is_null_tids;
+
+	slot = table_slot_create(toastrel, NULL);
+
+	if (!table_tuple_fetch_row_version(toastrel, tid, snapshot, slot))
+	{
+		ExecDropSingleTupleTableSlot(slot);
+		return;
+	}
+
+	tids_datum = slot_getattr(slot, 4, &is_null_tids);
+
+	if (!is_null_tids)
+	{
+		ArrayType  *arr = DatumGetArrayTypePCopy(tids_datum);
+		Datum	   *elems;
+		bool	   *nulls;
+		int			nelems;
+		int			i;
+
+		deconstruct_array_builtin(arr, TIDOID, &elems, &nulls, &nelems);
+
+		ExecClearTuple(slot);
+
+		for (i = 0; i < nelems; i++)
+		{
+			ItemPointer elem_tid = (ItemPointer) DatumGetPointer(elems[i]);
+			if (!ItemPointerEquals(elem_tid, tid))
+			{
+				toast_delete_datum_direct_recursive(toastrel, elem_tid, is_speculative);
+			}
+		}
+		pfree(elems);
+		pfree(nulls);
+		pfree(arr);
+	}
+	else
+	{
+		ExecClearTuple(slot);
+	}
+
+	ExecDropSingleTupleTableSlot(slot);
+
+	if (is_speculative)
+		heap_abort_speculative(toastrel, tid);
+	else
+		simple_heap_delete(toastrel, tid);
+}
diff --git a/src/backend/access/table/toast_helper.c b/src/backend/access/table/toast_helper.c
index 2613d9dc095..d526b562abe 100644
--- a/src/backend/access/table/toast_helper.c
+++ b/src/backend/access/table/toast_helper.c
@@ -74,10 +74,10 @@ toast_tuple_init(ToastTupleContext *ttc)
 			 * or oid8, and these can have different sizes.
 			 */
 			if (att->attlen == -1 && !ttc->ttc_oldisnull[i] &&
-				VARATT_IS_EXTERNAL_ONDISK(old_value))
+				(VARATT_IS_EXTERNAL_ONDISK(old_value) || VARATT_IS_EXTERNAL_DIRECT(old_value)))
 			{
 				if (ttc->ttc_isnull[i] ||
-					!VARATT_IS_EXTERNAL_ONDISK(new_value) ||
+					!(VARATT_IS_EXTERNAL_ONDISK(new_value) || VARATT_IS_EXTERNAL_DIRECT(new_value)) ||
 					VARTAG_EXTERNAL(old_value) != VARTAG_EXTERNAL(new_value) ||
 					memcmp(old_value, new_value,
 						   VARSIZE_EXTERNAL(old_value)) != 0)
@@ -346,7 +346,7 @@ toast_delete_external(Relation rel, const Datum *values, const bool *isnull,
 
 			if (isnull[i])
 				continue;
-			else if (VARATT_IS_EXTERNAL_ONDISK(DatumGetPointer(value)))
+			else if (VARATT_IS_EXTERNAL_ONDISK(DatumGetPointer(value)) || VARATT_IS_EXTERNAL_DIRECT(DatumGetPointer(value)))
 				toast_delete_datum(rel, value, is_speculative);
 		}
 	}
diff --git a/src/backend/utils/adt/arrayfuncs.c b/src/backend/utils/adt/arrayfuncs.c
index ef66182b047..0067866d7e0 100644
--- a/src/backend/utils/adt/arrayfuncs.c
+++ b/src/backend/utils/adt/arrayfuncs.c
@@ -3740,6 +3740,12 @@ deconstruct_array_builtin(const ArrayType *array,
 			elmalign = TYPALIGN_INT;
 			break;
 
+		case INT8OID:
+			elmlen = sizeof(int64);
+			elmbyval = true;
+			elmalign = TYPALIGN_DOUBLE;
+			break;
+
 		case OIDOID:
 			elmlen = sizeof(Oid);
 			elmbyval = true;
diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat
index c57441f7d98..9322bf0c36f 100644
--- a/src/backend/utils/misc/guc_parameters.dat
+++ b/src/backend/utils/misc/guc_parameters.dat
@@ -3120,6 +3120,13 @@
   show_hook => 'show_timing_clock_source',
 },
 
+{ name => 'toast_flavour', type => 'enum', context => 'PGC_USERSET', group => 'CLIENT_CONN_STATEMENT',
+  short_desc => 'Sets the TOAST flavour to use for new writes.',
+  variable => 'toast_flavour',
+  boot_val => 'TOAST_FLAVOUR_PLAIN',
+  options => 'toast_flavour_options',
+},
+
 { name => 'trace_connection_negotiation', type => 'bool', context => 'PGC_POSTMASTER', group => 'DEVELOPER_OPTIONS',
   short_desc => 'Logs details of pre-authentication connection handshake.',
   flags => 'GUC_NOT_IN_SAMPLE',
diff --git a/src/include/access/detoast.h b/src/include/access/detoast.h
index 93b7a253760..4ec7eff2a37 100644
--- a/src/include/access/detoast.h
+++ b/src/include/access/detoast.h
@@ -29,12 +29,23 @@ do { \
 	memcpy(&(toast_pointer), VARDATA_EXTERNAL(attre), sizeof(toast_pointer)); \
 } while (0)
 
+#define VARATT_EXTERNAL_GET_POINTER_DIRECT(toast_pointer, attr) \
+do { \
+	varattrib_1b_e *attre = (varattrib_1b_e *) (attr); \
+	Assert(VARATT_IS_EXTERNAL(attre)); \
+	Assert(VARTAG_EXTERNAL(attre) == VARTAG_DIRECT); \
+	Assert(VARSIZE_EXTERNAL(attre) == sizeof(toast_pointer) + VARHDRSZ_EXTERNAL); \
+	memcpy(&(toast_pointer), VARDATA_EXTERNAL(attre), sizeof(toast_pointer)); \
+} while (0)
+
 /* Size of an EXTERNAL datum that contains a standard TOAST pointer */
 #define TOAST_OID_POINTER_SIZE (VARHDRSZ_EXTERNAL + sizeof(varatt_external_oid))
 
 /* Size of an EXTERNAL datum that contains an Oid8 TOAST pointer */
 #define TOAST_OID8_POINTER_SIZE (VARHDRSZ_EXTERNAL + sizeof(varatt_external_oid8))
 
+#define DIRECT_POINTER_SIZE (VARHDRSZ_EXTERNAL + sizeof(varatt_direct))
+
 /* Size of an EXTERNAL datum that contains an indirection pointer */
 #define INDIRECT_POINTER_SIZE (VARHDRSZ_EXTERNAL + sizeof(varatt_indirect))
 
diff --git a/src/test/regress/expected/cluster.out b/src/test/regress/expected/cluster.out
index 64a9c35fcd0..4c2506ed7cf 100644
--- a/src/test/regress/expected/cluster.out
+++ b/src/test/regress/expected/cluster.out
@@ -308,6 +308,7 @@ WHERE pg_class.oid=indexrelid
 
 -- Verify that toast tables are clusterable
 CLUSTER pg_toast.pg_toast_826 USING pg_toast_826_index;
+ERROR:  cannot cluster on partial index "pg_toast_826_index"
 -- Verify that clustering all tables does in fact cluster the right ones
 CREATE USER regress_clstr_user;
 CREATE TABLE clstr_1 (a INT PRIMARY KEY);
diff --git a/src/test/regress/expected/direct_toast.out b/src/test/regress/expected/direct_toast.out
new file mode 100644
index 00000000000..489bb84b7d0
--- /dev/null
+++ b/src/test/regress/expected/direct_toast.out
@@ -0,0 +1,565 @@
+--
+-- Tests for direct TOAST flavour
+--
+SET toast_flavour = 'direct';
+-- Check GUC
+SHOW toast_flavour;
+ toast_flavour 
+---------------
+ direct
+(1 row)
+
+CREATE TABLE dirtoasttest(descr text, f1 text);
+ALTER TABLE dirtoasttest ALTER COLUMN f1 SET STORAGE EXTERNAL;
+-- Single-chunk toast (or small multi-chunk)
+INSERT INTO dirtoasttest VALUES ('toasted-1', repeat('1234567890', 1000)); -- 10KB (uncompressed, so ~5 chunks)
+-- Multi-chunk toast
+INSERT INTO dirtoasttest VALUES ('toasted-multi', repeat('1234567890', 5000)); -- 50KB (uncompressed, so ~25 chunks)
+REINDEX TABLE dirtoasttest;
+-- Verify toast table structure and contents
+DO $$
+DECLARE
+    toast_relname text;
+    r record;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'dirtoasttest');
+    IF toast_relname IS NOT NULL THEN
+        FOR r IN EXECUTE 'SELECT chunk_id IS NULL as id_isnull, chunk_seq, chunk_tids IS NULL as tids_isnull FROM ' || toast_relname || ' ORDER BY id_isnull desc, chunk_id, chunk_seq' LOOP
+            RAISE NOTICE 'chunk: id_isnull=%, seq=%, tids_isnull=%', r.id_isnull, r.chunk_seq, r.tids_isnull;
+        END LOOP;
+    ELSE
+        RAISE NOTICE 'no toast table';
+    END IF;
+END$$;
+NOTICE:  chunk: id_isnull=t, seq=0, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=0, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=1, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=1, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=2, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=2, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=3, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=3, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=4, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=4, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=5, tids_isnull=f
+NOTICE:  chunk: id_isnull=t, seq=5, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=6, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=7, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=8, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=9, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=10, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=11, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=12, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=13, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=14, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=15, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=16, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=17, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=18, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=19, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=20, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=21, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=22, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=23, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=24, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=25, tids_isnull=f
+-- Read only descr (should work)
+SELECT descr FROM dirtoasttest;
+     descr     
+---------------
+ toasted-1
+ toasted-multi
+(2 rows)
+
+-- Read f1 IS NULL (should work, and return false)
+SELECT descr, f1 IS NULL FROM dirtoasttest;
+     descr     | ?column? 
+---------------+----------
+ toasted-1     | f
+ toasted-multi | f
+(2 rows)
+
+-- Read values while GUC is still 'direct'
+SELECT descr, length(f1), substring(f1, 1, 10), substring(f1, length(f1)-9, 10) FROM dirtoasttest;
+     descr     | length | substring  | substring  
+---------------+--------+------------+------------
+ toasted-1     |  10000 | 1234567890 | 1234567890
+ toasted-multi |  50000 | 1234567890 | 1234567890
+(2 rows)
+
+-- Reset GUC to plain and try reading (should still work because read path is automatic)
+SET toast_flavour = 'plain';
+SHOW toast_flavour;
+ toast_flavour 
+---------------
+ plain
+(1 row)
+
+SELECT descr, length(f1), substring(f1, 1, 10), substring(f1, length(f1)-9, 10) FROM dirtoasttest;
+     descr     | length | substring  | substring  
+---------------+--------+------------+------------
+ toasted-1     |  10000 | 1234567890 | 1234567890
+ toasted-multi |  50000 | 1234567890 | 1234567890
+(2 rows)
+
+-- Test slice reading
+SELECT descr, substring(f1, 500, 20) FROM dirtoasttest WHERE descr = 'toasted-multi';
+     descr     |      substring       
+---------------+----------------------
+ toasted-multi | 01234567890123456789
+(1 row)
+
+SELECT descr, substring(f1, 45000, 20) FROM dirtoasttest WHERE descr = 'toasted-multi';
+     descr     |      substring       
+---------------+----------------------
+ toasted-multi | 01234567890123456789
+(1 row)
+
+-- Test update (should write as 'plain' now because GUC is 'plain')
+-- We use a smaller value to avoid too many chunks, but still toasted.
+-- Actually, updated value will also be toasted if it's large.
+-- 'toasted-1' is 10KB. f1 || 'edited' is 10006 bytes. It will be toasted.
+UPDATE dirtoasttest SET f1 = f1 || 'edited' WHERE descr = 'toasted-1';
+SELECT descr, length(f1), substring(f1, length(f1)-9, 10) FROM dirtoasttest WHERE descr = 'toasted-1';
+   descr   | length | substring  
+-----------+--------+------------
+ toasted-1 |  10006 | 7890edited
+(1 row)
+
+-- Toast table should now contain some 'plain' toast (no tid array) and some 'direct' toast.
+-- The updated 'toasted-1' should be plain.
+-- Let's check toast table again.
+DO $$
+DECLARE
+    toast_relname text;
+    r record;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'dirtoasttest');
+    IF toast_relname IS NOT NULL THEN
+        FOR r IN EXECUTE 'SELECT chunk_id IS NULL as id_isnull, chunk_seq, chunk_tids IS NULL as tids_isnull FROM ' || toast_relname || ' ORDER BY id_isnull desc, chunk_id, chunk_seq' LOOP
+            RAISE NOTICE 'chunk: id_isnull=%, seq=%, tids_isnull=%', r.id_isnull, r.chunk_seq, r.tids_isnull;
+        END LOOP;
+    END IF;
+END$$;
+NOTICE:  chunk: id_isnull=t, seq=0, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=1, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=2, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=3, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=4, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=5, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=6, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=7, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=8, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=9, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=10, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=11, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=12, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=13, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=14, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=15, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=16, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=17, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=18, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=19, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=20, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=21, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=22, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=23, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=24, tids_isnull=t
+NOTICE:  chunk: id_isnull=t, seq=25, tids_isnull=f
+NOTICE:  chunk: id_isnull=f, seq=0, tids_isnull=t
+NOTICE:  chunk: id_isnull=f, seq=1, tids_isnull=t
+NOTICE:  chunk: id_isnull=f, seq=2, tids_isnull=t
+NOTICE:  chunk: id_isnull=f, seq=3, tids_isnull=t
+NOTICE:  chunk: id_isnull=f, seq=4, tids_isnull=t
+NOTICE:  chunk: id_isnull=f, seq=5, tids_isnull=t
+-- Delete and vacuum
+DELETE FROM dirtoasttest;
+VACUUM dirtoasttest;
+-- Toast table should be empty
+DO $$
+DECLARE
+    toast_relname text;
+    cnt int;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'dirtoasttest');
+    IF toast_relname IS NOT NULL THEN
+        EXECUTE 'SELECT count(*) FROM ' || toast_relname INTO cnt;
+        RAISE NOTICE 'toast table row count: %', cnt;
+    ELSE
+        RAISE NOTICE 'no toast table';
+    END IF;
+END$$;
+NOTICE:  toast table row count: 0
+-- Verify index skip and InvalidOid usage
+-- We insert two direct toast values. They should both get chunk_id = NULL.
+-- Since the index is partial (WHERE chunk_id IS NOT NULL), they won't be indexed,
+-- and thus won't conflict on the unique index.
+SET toast_flavour = 'direct';
+INSERT INTO dirtoasttest VALUES ('toasted-idx-1', repeat('a', 3000));
+INSERT INTO dirtoasttest VALUES ('toasted-idx-2', repeat('b', 3000));
+-- Should succeed.
+-- Verify they have chunk_id = NULL
+DO $$
+DECLARE
+    toast_relname text;
+    r record;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'dirtoasttest');
+    FOR r IN EXECUTE 'SELECT chunk_id IS NULL as is_direct, chunk_seq, chunk_tids IS NULL as tids_isnull FROM ' || toast_relname || ' ORDER BY chunk_seq, tids_isnull' LOOP
+        RAISE NOTICE 'dirtoasttest chunk: is_direct=%, seq=%, tids_isnull=%', r.is_direct, r.chunk_seq, r.tids_isnull;
+    END LOOP;
+END$$;
+NOTICE:  dirtoasttest chunk: is_direct=t, seq=0, tids_isnull=t
+NOTICE:  dirtoasttest chunk: is_direct=t, seq=0, tids_isnull=t
+NOTICE:  dirtoasttest chunk: is_direct=t, seq=1, tids_isnull=f
+NOTICE:  dirtoasttest chunk: is_direct=t, seq=1, tids_isnull=f
+REINDEX TABLE dirtoasttest;
+DROP TABLE dirtoasttest;
+-- Test Table Storage Parameter 'toast_flavour'
+SET toast_flavour = 'plain'; -- GUC is plain
+-- 1. Table option 'direct'
+CREATE TABLE tab_direct(descr text, f1 text) WITH (toast_flavour = 'direct');
+ALTER TABLE tab_direct ALTER COLUMN f1 SET STORAGE EXTERNAL;
+INSERT INTO tab_direct VALUES ('opt-direct', repeat('d', 3000));
+-- Verify it is direct (chunk_tids is not null, and chunk_id is NULL)
+DO $$
+DECLARE
+    toast_relname text;
+    r record;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'tab_direct');
+    FOR r IN EXECUTE 'SELECT chunk_id IS NULL as is_direct, chunk_tids IS NULL as tids_isnull FROM ' || toast_relname LOOP
+        RAISE NOTICE 'tab_direct chunk: is_direct=%, tids_isnull=%', r.is_direct, r.tids_isnull;
+    END LOOP;
+END$$;
+NOTICE:  tab_direct chunk: is_direct=t, tids_isnull=t
+NOTICE:  tab_direct chunk: is_direct=t, tids_isnull=f
+-- 2. Table option 'plain', GUC is 'direct'
+SET toast_flavour = 'direct';
+CREATE TABLE tab_plain(descr text, f1 text) WITH (toast_flavour = 'plain');
+ALTER TABLE tab_plain ALTER COLUMN f1 SET STORAGE EXTERNAL;
+INSERT INTO tab_plain VALUES ('opt-plain', repeat('p', 3000));
+-- Verify it is plain (chunk_tids is null, chunk_id is NOT NULL)
+DO $$
+DECLARE
+    toast_relname text;
+    r record;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'tab_plain');
+    FOR r IN EXECUTE 'SELECT chunk_id IS NULL as is_direct, chunk_tids IS NULL as tids_isnull FROM ' || toast_relname LOOP
+        RAISE NOTICE 'tab_plain chunk: is_direct=%, tids_isnull=%', r.is_direct, r.tids_isnull;
+    END LOOP;
+END$$;
+NOTICE:  tab_plain chunk: is_direct=f, tids_isnull=t
+NOTICE:  tab_plain chunk: is_direct=f, tids_isnull=t
+-- 3. Default (no option), follows GUC
+CREATE TABLE tab_default(descr text, f1 text);
+ALTER TABLE tab_default ALTER COLUMN f1 SET STORAGE EXTERNAL;
+-- GUC is direct -> writes direct (chunk_id = NULL)
+INSERT INTO tab_default VALUES ('default-direct', repeat('g', 3000));
+-- GUC is plain -> writes plain (chunk_id <> NULL)
+SET toast_flavour = 'plain';
+INSERT INTO tab_default VALUES ('default-plain', repeat('h', 3000));
+-- Verify contents
+DO $$
+DECLARE
+    toast_relname text;
+    r record;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'tab_default');
+    FOR r IN EXECUTE 'SELECT chunk_id IS NULL as is_direct, chunk_tids IS NULL as tids_isnull FROM ' || toast_relname || ' ORDER BY is_direct desc, tids_isnull' LOOP
+        RAISE NOTICE 'tab_default chunk: is_direct=%, tids_isnull=%', r.is_direct, r.tids_isnull;
+    END LOOP;
+END$$;
+NOTICE:  tab_default chunk: is_direct=t, tids_isnull=f
+NOTICE:  tab_default chunk: is_direct=t, tids_isnull=t
+NOTICE:  tab_default chunk: is_direct=f, tids_isnull=t
+NOTICE:  tab_default chunk: is_direct=f, tids_isnull=t
+-- 4. Alter table SET toast_flavour
+ALTER TABLE tab_default SET (toast_flavour = 'direct');
+-- GUC is plain -> should write direct because of table option (chunk_id = NULL)
+INSERT INTO tab_default VALUES ('default-altered-direct', repeat('i', 3000));
+-- 5. Alter table RESET toast_flavour
+ALTER TABLE tab_default RESET (toast_flavour);
+-- GUC is plain -> should write plain (chunk_id <> NULL)
+INSERT INTO tab_default VALUES ('default-reset-plain', repeat('j', 3000));
+-- Verify after alters
+DO $$
+DECLARE
+    toast_relname text;
+    r record;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'tab_default');
+    FOR r IN EXECUTE 'SELECT chunk_id IS NULL as is_direct, chunk_tids IS NULL as tids_isnull FROM ' || toast_relname || ' ORDER BY is_direct desc, tids_isnull' LOOP
+        RAISE NOTICE 'tab_default altered chunk: is_direct=%, tids_isnull=%', r.is_direct, r.tids_isnull;
+    END LOOP;
+END$$;
+NOTICE:  tab_default altered chunk: is_direct=t, tids_isnull=f
+NOTICE:  tab_default altered chunk: is_direct=t, tids_isnull=f
+NOTICE:  tab_default altered chunk: is_direct=t, tids_isnull=t
+NOTICE:  tab_default altered chunk: is_direct=t, tids_isnull=t
+NOTICE:  tab_default altered chunk: is_direct=f, tids_isnull=t
+NOTICE:  tab_default altered chunk: is_direct=f, tids_isnull=t
+NOTICE:  tab_default altered chunk: is_direct=f, tids_isnull=t
+NOTICE:  tab_default altered chunk: is_direct=f, tids_isnull=t
+-- Clean up
+DROP TABLE tab_direct;
+DROP TABLE tab_plain;
+DROP TABLE tab_default;
+--
+-- Test Recursive Tree Direct TOAST (>100 chunks, with chunk_tid_offsets)
+--
+CREATE TABLE tab_tree(descr text, f1 text) WITH (toast_flavour = 'direct');
+ALTER TABLE tab_tree ALTER COLUMN f1 SET STORAGE EXTERNAL;
+-- 241,200 bytes (~120 chunks > 100 threshold -> 120 leaf chunks, 3 level-1 nodes, 1 root node)
+INSERT INTO tab_tree VALUES ('tree-toast-1', repeat('abcdefghijklmnopqrstuvwxyz0123456789', 6700));
+-- Verify table length and checksum
+SELECT descr, length(f1), md5(f1) = md5(repeat('abcdefghijklmnopqrstuvwxyz0123456789', 6700)) as md5_match FROM tab_tree;
+    descr     | length | md5_match 
+--------------+--------+-----------
+ tree-toast-1 | 241200 | t
+(1 row)
+
+-- Verify slices: start, middle crossing chunk/node boundaries, end
+SELECT descr, substring(f1, 1, 36) FROM tab_tree;
+    descr     |              substring               
+--------------+--------------------------------------
+ tree-toast-1 | abcdefghijklmnopqrstuvwxyz0123456789
+(1 row)
+
+SELECT descr, substring(f1, 1990, 36) FROM tab_tree;
+    descr     |              substring               
+--------------+--------------------------------------
+ tree-toast-1 | jklmnopqrstuvwxyz0123456789abcdefghi
+(1 row)
+
+SELECT descr, substring(f1, 99990, 36) FROM tab_tree;
+    descr     |              substring               
+--------------+--------------------------------------
+ tree-toast-1 | rstuvwxyz0123456789abcdefghijklmnopq
+(1 row)
+
+SELECT descr, substring(f1, 241165, 36) FROM tab_tree;
+    descr     |              substring               
+--------------+--------------------------------------
+ tree-toast-1 | abcdefghijklmnopqrstuvwxyz0123456789
+(1 row)
+
+-- Inspect toast table structure for tree nodes
+DO $$
+DECLARE
+    toast_relname text;
+    r record;
+    leaf_count int := 0;
+    node_count int := 0;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'tab_tree');
+    FOR r IN EXECUTE 'SELECT chunk_id IS NULL as id_null, chunk_data IS NULL as data_null, array_length(chunk_tids, 1) as num_tids, array_length(chunk_tid_offsets, 1) as num_offsets FROM ' || toast_relname || ' ORDER BY chunk_seq' LOOP
+        IF r.data_null THEN
+            node_count := node_count + 1;
+            IF r.num_offsets <> r.num_tids + 1 THEN
+                RAISE EXCEPTION 'offset count % does not match tid count + 1 (%)', r.num_offsets, r.num_tids + 1;
+            END IF;
+        ELSE
+            leaf_count := leaf_count + 1;
+        END IF;
+    END LOOP;
+    RAISE NOTICE 'tree toast structure: leaf_count=%, node_count=%', leaf_count, node_count;
+END$$;
+NOTICE:  tree toast structure: leaf_count=121, node_count=4
+-- Verify root node offsets span from 0 to full length
+DO $$
+DECLARE
+    toast_relname text;
+    root_offsets bigint[];
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'tab_tree');
+    EXECUTE 'SELECT chunk_tid_offsets FROM ' || toast_relname || ' WHERE chunk_data IS NULL ORDER BY chunk_seq DESC LIMIT 1' INTO root_offsets;
+    RAISE NOTICE 'root offsets: first=%, last=%', root_offsets[1], root_offsets[array_length(root_offsets, 1)];
+END$$;
+NOTICE:  root offsets: first=0, last=241200
+-- Test update with tree toast
+UPDATE tab_tree SET f1 = f1 || '_updated';
+SELECT descr, length(f1), substring(f1, 241200, 9) FROM tab_tree;
+    descr     | length | substring 
+--------------+--------+-----------
+ tree-toast-1 | 241208 | 9_updated
+(1 row)
+
+-- Delete and vacuum
+DELETE FROM tab_tree;
+VACUUM tab_tree;
+DO $$
+DECLARE
+    toast_relname text;
+    cnt int;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'tab_tree');
+    EXECUTE 'SELECT count(*) FROM ' || toast_relname INTO cnt;
+    RAISE NOTICE 'tree toast table count after vacuum: %', cnt;
+END$$;
+NOTICE:  tree toast table count after vacuum: 0
+DROP TABLE tab_tree;
+--
+-- Test Compression and Chunk ID Introspection on Direct TOAST
+--
+CREATE TABLE tab_intro_plain(descr text, f text) WITH (toast_flavour = 'plain');
+CREATE TABLE tab_intro_direct(descr text, f text) WITH (toast_flavour = 'direct');
+INSERT INTO tab_intro_plain SELECT 'uncompressed-external-plain', string_agg(md5(i::text), '') FROM generate_series(1, 200) i;
+INSERT INTO tab_intro_direct SELECT 'uncompressed-external-direct', string_agg(md5(i::text), '') FROM generate_series(1, 200) i;
+INSERT INTO tab_intro_plain VALUES ('compressed-external-plain', repeat('abcdefghijklmnopqrstuvwxyz0123456789', 5000));
+INSERT INTO tab_intro_direct VALUES ('compressed-external-direct', repeat('abcdefghijklmnopqrstuvwxyz0123456789', 5000));
+SELECT p.descr, pg_column_compression(p.f) AS plain_comp, pg_column_toast_chunk_id(p.f) IS NOT NULL AS plain_has_chunk_id
+FROM tab_intro_plain p
+ORDER BY p.descr;
+            descr            | plain_comp | plain_has_chunk_id 
+-----------------------------+------------+--------------------
+ compressed-external-plain   | pglz       | t
+ uncompressed-external-plain |            | t
+(2 rows)
+
+SELECT d.descr, pg_column_compression(d.f) AS direct_comp, pg_column_toast_chunk_id(d.f) AS direct_chunk_id
+FROM tab_intro_direct d
+ORDER BY d.descr;
+            descr             | direct_comp | direct_chunk_id 
+------------------------------+-------------+-----------------
+ compressed-external-direct   | pglz        |                
+ uncompressed-external-direct |             |                
+(2 rows)
+
+DROP TABLE tab_intro_plain;
+DROP TABLE tab_intro_direct;
+--
+-- Test Partitioned Tables with Mixed Toast Flavours and Cross-Partition Updates
+--
+CREATE TABLE part_toast(id int, val text) PARTITION BY RANGE (id);
+CREATE TABLE part_toast_p1 PARTITION OF part_toast FOR VALUES FROM (1) TO (100) WITH (toast_flavour = 'direct');
+CREATE TABLE part_toast_p2 PARTITION OF part_toast FOR VALUES FROM (100) TO (200) WITH (toast_flavour = 'plain');
+INSERT INTO part_toast SELECT 1, string_agg(md5(i::text), '') FROM generate_series(1, 200) i;
+INSERT INTO part_toast SELECT 101, string_agg(md5(i::text), '') FROM generate_series(1, 200) i;
+SELECT id, length(val), substring(val, 1, 10), pg_column_toast_chunk_id(val) IS NOT NULL AS has_chunk_id
+FROM part_toast
+ORDER BY id;
+ id  | length | substring  | has_chunk_id 
+-----+--------+------------+--------------
+   1 |   6400 | c4ca4238a0 | f
+ 101 |   6400 | c4ca4238a0 | t
+(2 rows)
+
+-- Move row from direct partition to plain partition
+UPDATE part_toast SET id = 102 WHERE id = 1;
+SELECT id, length(val), substring(val, 1, 10), pg_column_toast_chunk_id(val) IS NOT NULL AS has_chunk_id
+FROM part_toast
+ORDER BY id;
+ id  | length | substring  | has_chunk_id 
+-----+--------+------------+--------------
+ 101 |   6400 | c4ca4238a0 | t
+ 102 |   6400 | c4ca4238a0 | t
+(2 rows)
+
+-- Move row from plain partition to direct partition
+UPDATE part_toast SET id = 2 WHERE id = 101;
+SELECT id, length(val), substring(val, 1, 10), pg_column_toast_chunk_id(val) IS NOT NULL AS has_chunk_id
+FROM part_toast
+ORDER BY id;
+ id  | length | substring  | has_chunk_id 
+-----+--------+------------+--------------
+   2 |   6400 | c4ca4238a0 | f
+ 102 |   6400 | c4ca4238a0 | t
+(2 rows)
+
+DROP TABLE part_toast;
+--
+-- Test Expression / Functional Indexes on Direct TOAST Columns
+--
+CREATE TABLE tab_expr_idx(id int primary key, payload text) WITH (toast_flavour = 'direct');
+CREATE INDEX idx_tab_expr_md5 ON tab_expr_idx (md5(payload));
+CREATE INDEX idx_tab_expr_substr ON tab_expr_idx (substring(payload, 1, 20));
+INSERT INTO tab_expr_idx VALUES (1, repeat('expr-index-test-payload-', 500));
+INSERT INTO tab_expr_idx VALUES (2, repeat('other-index-test-payload-', 500));
+SET enable_seqscan = off;
+SELECT id, length(payload) FROM tab_expr_idx WHERE md5(payload) = md5(repeat('expr-index-test-payload-', 500));
+ id | length 
+----+--------
+  1 |  12000
+(1 row)
+
+SELECT id, length(payload) FROM tab_expr_idx WHERE substring(payload, 1, 20) = 'expr-index-test-payl';
+ id | length 
+----+--------
+  1 |  12000
+(1 row)
+
+RESET enable_seqscan;
+--
+-- Test Table Maintenance and Rewrites (VACUUM FULL, CLUSTER, ALTER TYPE, TRUNCATE)
+--
+VACUUM FULL tab_expr_idx;
+SELECT id, length(payload), substring(payload, 1, 24) FROM tab_expr_idx ORDER BY id;
+ id | length |        substring         
+----+--------+--------------------------
+  1 |  12000 | expr-index-test-payload-
+  2 |  12500 | other-index-test-payload
+(2 rows)
+
+CLUSTER tab_expr_idx USING tab_expr_idx_pkey;
+SELECT id, length(payload), substring(payload, 1, 24) FROM tab_expr_idx ORDER BY id;
+ id | length |        substring         
+----+--------+--------------------------
+  1 |  12000 | expr-index-test-payload-
+  2 |  12500 | other-index-test-payload
+(2 rows)
+
+ALTER TABLE tab_expr_idx ALTER COLUMN payload TYPE varchar(20000);
+SELECT id, length(payload), substring(payload, 1, 24) FROM tab_expr_idx ORDER BY id;
+ id | length |        substring         
+----+--------+--------------------------
+  1 |  12000 | expr-index-test-payload-
+  2 |  12500 | other-index-test-payload
+(2 rows)
+
+TRUNCATE tab_expr_idx;
+SELECT count(*) FROM tab_expr_idx;
+ count 
+-------
+     0
+(1 row)
+
+DROP TABLE tab_expr_idx;
+--
+-- Test VACUUM FULL / CLUSTER restrictions on direct TOAST tables
+--
+CREATE TABLE tab_toast_maint(id int, val text) WITH (toast_flavour = 'direct');
+INSERT INTO tab_toast_maint VALUES (1, repeat('maint-test-', 500));
+-- VACUUM FULL on the parent table succeeds and rebuilds direct toast safely
+VACUUM FULL tab_toast_maint;
+SELECT id, length(val) FROM tab_toast_maint;
+ id | length 
+----+--------
+  1 |   5500
+(1 row)
+
+-- CLUSTER on direct TOAST table directly is rejected
+DO $$
+DECLARE
+    toast_relname text;
+    toast_idxname text;
+BEGIN
+    SELECT c2.relname, c3.relname INTO toast_relname, toast_idxname
+    FROM pg_class c1
+    JOIN pg_class c2 ON c1.reltoastrelid = c2.oid
+    JOIN pg_index i ON c2.oid = i.indrelid
+    JOIN pg_class c3 ON i.indexrelid = c3.oid
+    WHERE c1.relname = 'tab_toast_maint';
+
+    -- CLUSTER directly on direct TOAST table should be rejected
+    BEGIN
+        EXECUTE 'CLUSTER pg_toast.' || toast_relname || ' USING ' || toast_idxname;
+        RAISE EXCEPTION 'CLUSTER on direct TOAST table should have failed';
+    EXCEPTION WHEN feature_not_supported THEN
+        RAISE NOTICE 'expected error caught for CLUSTER on direct TOAST table: %', regexp_replace(SQLERRM, 'pg_toast_[0-9]+_index', 'pg_toast_xxx_index');
+    END;
+END$$;
+NOTICE:  expected error caught for CLUSTER on direct TOAST table: cannot cluster on partial index "pg_toast_xxx_index"
+DROP TABLE tab_toast_maint;
diff --git a/src/test/regress/expected/psql.out b/src/test/regress/expected/psql.out
index e8605dd041f..84578820198 100644
--- a/src/test/regress/expected/psql.out
+++ b/src/test/regress/expected/psql.out
@@ -5191,14 +5191,16 @@ drop role regress_partitioning_role;
 -- \d on toast table (use pg_statistic's toast table, which has a known name)
 \d pg_toast.pg_toast_2619
 TOAST table "pg_toast.pg_toast_2619"
-   Column   |  Type   
-------------+---------
- chunk_id   | oid
- chunk_seq  | integer
- chunk_data | bytea
+      Column       |   Type   
+-------------------+----------
+ chunk_id          | oid
+ chunk_seq         | integer
+ chunk_data        | bytea
+ chunk_tids        | tid[]
+ chunk_tid_offsets | bigint[]
 Owning table: "pg_catalog.pg_statistic"
 Indexes:
-    "pg_toast_2619_index" PRIMARY KEY, btree (chunk_id, chunk_seq)
+    "pg_toast_2619_index" PRIMARY KEY, btree (chunk_id, chunk_seq) WHERE chunk_id IS NOT NULL
 
 -- check printing info about access methods
 \dA
diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule
index 75063f87a4a..48b45f0f775 100644
--- a/src/test/regress/parallel_schedule
+++ b/src/test/regress/parallel_schedule
@@ -102,7 +102,7 @@ test: publication subscription
 # Another group of parallel tests
 # select_views depends on create_view
 # ----------
-test: select_views portals_p2 foreign_key dependency guc bitmapops combocid tsearch tsdicts foreign_data window xmlmap functional_deps advisory_lock indirect_toast equivclass stats_rewrite
+test: select_views portals_p2 foreign_key dependency guc bitmapops combocid tsearch tsdicts foreign_data window xmlmap functional_deps advisory_lock indirect_toast direct_toast equivclass stats_rewrite
 
 # ----------
 # Another group of parallel tests (JSON related)
diff --git a/src/test/regress/sql/direct_toast.sql b/src/test/regress/sql/direct_toast.sql
new file mode 100644
index 00000000000..7ab3f970e4c
--- /dev/null
+++ b/src/test/regress/sql/direct_toast.sql
@@ -0,0 +1,397 @@
+--
+-- Tests for direct TOAST flavour
+--
+
+SET toast_flavour = 'direct';
+
+-- Check GUC
+SHOW toast_flavour;
+
+CREATE TABLE dirtoasttest(descr text, f1 text);
+ALTER TABLE dirtoasttest ALTER COLUMN f1 SET STORAGE EXTERNAL;
+
+-- Single-chunk toast (or small multi-chunk)
+INSERT INTO dirtoasttest VALUES ('toasted-1', repeat('1234567890', 1000)); -- 10KB (uncompressed, so ~5 chunks)
+
+-- Multi-chunk toast
+INSERT INTO dirtoasttest VALUES ('toasted-multi', repeat('1234567890', 5000)); -- 50KB (uncompressed, so ~25 chunks)
+REINDEX TABLE dirtoasttest;
+
+-- Verify toast table structure and contents
+DO $$
+DECLARE
+    toast_relname text;
+    r record;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'dirtoasttest');
+    IF toast_relname IS NOT NULL THEN
+        FOR r IN EXECUTE 'SELECT chunk_id IS NULL as id_isnull, chunk_seq, chunk_tids IS NULL as tids_isnull FROM ' || toast_relname || ' ORDER BY id_isnull desc, chunk_id, chunk_seq' LOOP
+            RAISE NOTICE 'chunk: id_isnull=%, seq=%, tids_isnull=%', r.id_isnull, r.chunk_seq, r.tids_isnull;
+        END LOOP;
+    ELSE
+        RAISE NOTICE 'no toast table';
+    END IF;
+END$$;
+
+-- Read only descr (should work)
+SELECT descr FROM dirtoasttest;
+-- Read f1 IS NULL (should work, and return false)
+SELECT descr, f1 IS NULL FROM dirtoasttest;
+
+-- Read values while GUC is still 'direct'
+SELECT descr, length(f1), substring(f1, 1, 10), substring(f1, length(f1)-9, 10) FROM dirtoasttest;
+
+-- Reset GUC to plain and try reading (should still work because read path is automatic)
+SET toast_flavour = 'plain';
+SHOW toast_flavour;
+
+SELECT descr, length(f1), substring(f1, 1, 10), substring(f1, length(f1)-9, 10) FROM dirtoasttest;
+
+-- Test slice reading
+SELECT descr, substring(f1, 500, 20) FROM dirtoasttest WHERE descr = 'toasted-multi';
+SELECT descr, substring(f1, 45000, 20) FROM dirtoasttest WHERE descr = 'toasted-multi';
+
+-- Test update (should write as 'plain' now because GUC is 'plain')
+-- We use a smaller value to avoid too many chunks, but still toasted.
+-- Actually, updated value will also be toasted if it's large.
+-- 'toasted-1' is 10KB. f1 || 'edited' is 10006 bytes. It will be toasted.
+UPDATE dirtoasttest SET f1 = f1 || 'edited' WHERE descr = 'toasted-1';
+SELECT descr, length(f1), substring(f1, length(f1)-9, 10) FROM dirtoasttest WHERE descr = 'toasted-1';
+
+-- Toast table should now contain some 'plain' toast (no tid array) and some 'direct' toast.
+-- The updated 'toasted-1' should be plain.
+-- Let's check toast table again.
+DO $$
+DECLARE
+    toast_relname text;
+    r record;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'dirtoasttest');
+    IF toast_relname IS NOT NULL THEN
+        FOR r IN EXECUTE 'SELECT chunk_id IS NULL as id_isnull, chunk_seq, chunk_tids IS NULL as tids_isnull FROM ' || toast_relname || ' ORDER BY id_isnull desc, chunk_id, chunk_seq' LOOP
+            RAISE NOTICE 'chunk: id_isnull=%, seq=%, tids_isnull=%', r.id_isnull, r.chunk_seq, r.tids_isnull;
+        END LOOP;
+    END IF;
+END$$;
+
+-- Delete and vacuum
+DELETE FROM dirtoasttest;
+VACUUM dirtoasttest;
+
+-- Toast table should be empty
+DO $$
+DECLARE
+    toast_relname text;
+    cnt int;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'dirtoasttest');
+    IF toast_relname IS NOT NULL THEN
+        EXECUTE 'SELECT count(*) FROM ' || toast_relname INTO cnt;
+        RAISE NOTICE 'toast table row count: %', cnt;
+    ELSE
+        RAISE NOTICE 'no toast table';
+    END IF;
+END$$;
+
+-- Verify index skip and InvalidOid usage
+-- We insert two direct toast values. They should both get chunk_id = NULL.
+-- Since the index is partial (WHERE chunk_id IS NOT NULL), they won't be indexed,
+-- and thus won't conflict on the unique index.
+SET toast_flavour = 'direct';
+INSERT INTO dirtoasttest VALUES ('toasted-idx-1', repeat('a', 3000));
+INSERT INTO dirtoasttest VALUES ('toasted-idx-2', repeat('b', 3000));
+-- Should succeed.
+
+-- Verify they have chunk_id = NULL
+DO $$
+DECLARE
+    toast_relname text;
+    r record;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'dirtoasttest');
+    FOR r IN EXECUTE 'SELECT chunk_id IS NULL as is_direct, chunk_seq, chunk_tids IS NULL as tids_isnull FROM ' || toast_relname || ' ORDER BY chunk_seq, tids_isnull' LOOP
+        RAISE NOTICE 'dirtoasttest chunk: is_direct=%, seq=%, tids_isnull=%', r.is_direct, r.chunk_seq, r.tids_isnull;
+    END LOOP;
+END$$;
+
+REINDEX TABLE dirtoasttest;
+
+DROP TABLE dirtoasttest;
+
+-- Test Table Storage Parameter 'toast_flavour'
+SET toast_flavour = 'plain'; -- GUC is plain
+
+-- 1. Table option 'direct'
+CREATE TABLE tab_direct(descr text, f1 text) WITH (toast_flavour = 'direct');
+ALTER TABLE tab_direct ALTER COLUMN f1 SET STORAGE EXTERNAL;
+INSERT INTO tab_direct VALUES ('opt-direct', repeat('d', 3000));
+
+-- Verify it is direct (chunk_tids is not null, and chunk_id is NULL)
+DO $$
+DECLARE
+    toast_relname text;
+    r record;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'tab_direct');
+    FOR r IN EXECUTE 'SELECT chunk_id IS NULL as is_direct, chunk_tids IS NULL as tids_isnull FROM ' || toast_relname LOOP
+        RAISE NOTICE 'tab_direct chunk: is_direct=%, tids_isnull=%', r.is_direct, r.tids_isnull;
+    END LOOP;
+END$$;
+
+-- 2. Table option 'plain', GUC is 'direct'
+SET toast_flavour = 'direct';
+CREATE TABLE tab_plain(descr text, f1 text) WITH (toast_flavour = 'plain');
+ALTER TABLE tab_plain ALTER COLUMN f1 SET STORAGE EXTERNAL;
+INSERT INTO tab_plain VALUES ('opt-plain', repeat('p', 3000));
+
+-- Verify it is plain (chunk_tids is null, chunk_id is NOT NULL)
+DO $$
+DECLARE
+    toast_relname text;
+    r record;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'tab_plain');
+    FOR r IN EXECUTE 'SELECT chunk_id IS NULL as is_direct, chunk_tids IS NULL as tids_isnull FROM ' || toast_relname LOOP
+        RAISE NOTICE 'tab_plain chunk: is_direct=%, tids_isnull=%', r.is_direct, r.tids_isnull;
+    END LOOP;
+END$$;
+
+-- 3. Default (no option), follows GUC
+CREATE TABLE tab_default(descr text, f1 text);
+ALTER TABLE tab_default ALTER COLUMN f1 SET STORAGE EXTERNAL;
+
+-- GUC is direct -> writes direct (chunk_id = NULL)
+INSERT INTO tab_default VALUES ('default-direct', repeat('g', 3000));
+
+-- GUC is plain -> writes plain (chunk_id <> NULL)
+SET toast_flavour = 'plain';
+INSERT INTO tab_default VALUES ('default-plain', repeat('h', 3000));
+
+-- Verify contents
+DO $$
+DECLARE
+    toast_relname text;
+    r record;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'tab_default');
+    FOR r IN EXECUTE 'SELECT chunk_id IS NULL as is_direct, chunk_tids IS NULL as tids_isnull FROM ' || toast_relname || ' ORDER BY is_direct desc, tids_isnull' LOOP
+        RAISE NOTICE 'tab_default chunk: is_direct=%, tids_isnull=%', r.is_direct, r.tids_isnull;
+    END LOOP;
+END$$;
+
+-- 4. Alter table SET toast_flavour
+ALTER TABLE tab_default SET (toast_flavour = 'direct');
+-- GUC is plain -> should write direct because of table option (chunk_id = NULL)
+INSERT INTO tab_default VALUES ('default-altered-direct', repeat('i', 3000));
+
+-- 5. Alter table RESET toast_flavour
+ALTER TABLE tab_default RESET (toast_flavour);
+-- GUC is plain -> should write plain (chunk_id <> NULL)
+INSERT INTO tab_default VALUES ('default-reset-plain', repeat('j', 3000));
+
+-- Verify after alters
+DO $$
+DECLARE
+    toast_relname text;
+    r record;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'tab_default');
+    FOR r IN EXECUTE 'SELECT chunk_id IS NULL as is_direct, chunk_tids IS NULL as tids_isnull FROM ' || toast_relname || ' ORDER BY is_direct desc, tids_isnull' LOOP
+        RAISE NOTICE 'tab_default altered chunk: is_direct=%, tids_isnull=%', r.is_direct, r.tids_isnull;
+    END LOOP;
+END$$;
+
+-- Clean up
+DROP TABLE tab_direct;
+DROP TABLE tab_plain;
+DROP TABLE tab_default;
+
+--
+-- Test Recursive Tree Direct TOAST (>100 chunks, with chunk_tid_offsets)
+--
+CREATE TABLE tab_tree(descr text, f1 text) WITH (toast_flavour = 'direct');
+ALTER TABLE tab_tree ALTER COLUMN f1 SET STORAGE EXTERNAL;
+
+-- 241,200 bytes (~120 chunks > 100 threshold -> 120 leaf chunks, 3 level-1 nodes, 1 root node)
+INSERT INTO tab_tree VALUES ('tree-toast-1', repeat('abcdefghijklmnopqrstuvwxyz0123456789', 6700));
+
+-- Verify table length and checksum
+SELECT descr, length(f1), md5(f1) = md5(repeat('abcdefghijklmnopqrstuvwxyz0123456789', 6700)) as md5_match FROM tab_tree;
+
+-- Verify slices: start, middle crossing chunk/node boundaries, end
+SELECT descr, substring(f1, 1, 36) FROM tab_tree;
+SELECT descr, substring(f1, 1990, 36) FROM tab_tree;
+SELECT descr, substring(f1, 99990, 36) FROM tab_tree;
+SELECT descr, substring(f1, 241165, 36) FROM tab_tree;
+
+-- Inspect toast table structure for tree nodes
+DO $$
+DECLARE
+    toast_relname text;
+    r record;
+    leaf_count int := 0;
+    node_count int := 0;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'tab_tree');
+    FOR r IN EXECUTE 'SELECT chunk_id IS NULL as id_null, chunk_data IS NULL as data_null, array_length(chunk_tids, 1) as num_tids, array_length(chunk_tid_offsets, 1) as num_offsets FROM ' || toast_relname || ' ORDER BY chunk_seq' LOOP
+        IF r.data_null THEN
+            node_count := node_count + 1;
+            IF r.num_offsets <> r.num_tids + 1 THEN
+                RAISE EXCEPTION 'offset count % does not match tid count + 1 (%)', r.num_offsets, r.num_tids + 1;
+            END IF;
+        ELSE
+            leaf_count := leaf_count + 1;
+        END IF;
+    END LOOP;
+    RAISE NOTICE 'tree toast structure: leaf_count=%, node_count=%', leaf_count, node_count;
+END$$;
+
+-- Verify root node offsets span from 0 to full length
+DO $$
+DECLARE
+    toast_relname text;
+    root_offsets bigint[];
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'tab_tree');
+    EXECUTE 'SELECT chunk_tid_offsets FROM ' || toast_relname || ' WHERE chunk_data IS NULL ORDER BY chunk_seq DESC LIMIT 1' INTO root_offsets;
+    RAISE NOTICE 'root offsets: first=%, last=%', root_offsets[1], root_offsets[array_length(root_offsets, 1)];
+END$$;
+
+-- Test update with tree toast
+UPDATE tab_tree SET f1 = f1 || '_updated';
+SELECT descr, length(f1), substring(f1, 241200, 9) FROM tab_tree;
+
+-- Delete and vacuum
+DELETE FROM tab_tree;
+VACUUM tab_tree;
+
+DO $$
+DECLARE
+    toast_relname text;
+    cnt int;
+BEGIN
+    SELECT 'pg_toast.' || relname INTO toast_relname FROM pg_class WHERE oid = (SELECT reltoastrelid FROM pg_class WHERE relname = 'tab_tree');
+    EXECUTE 'SELECT count(*) FROM ' || toast_relname INTO cnt;
+    RAISE NOTICE 'tree toast table count after vacuum: %', cnt;
+END$$;
+
+DROP TABLE tab_tree;
+
+--
+-- Test Compression and Chunk ID Introspection on Direct TOAST
+--
+CREATE TABLE tab_intro_plain(descr text, f text) WITH (toast_flavour = 'plain');
+CREATE TABLE tab_intro_direct(descr text, f text) WITH (toast_flavour = 'direct');
+
+INSERT INTO tab_intro_plain SELECT 'uncompressed-external-plain', string_agg(md5(i::text), '') FROM generate_series(1, 200) i;
+INSERT INTO tab_intro_direct SELECT 'uncompressed-external-direct', string_agg(md5(i::text), '') FROM generate_series(1, 200) i;
+
+INSERT INTO tab_intro_plain VALUES ('compressed-external-plain', repeat('abcdefghijklmnopqrstuvwxyz0123456789', 5000));
+INSERT INTO tab_intro_direct VALUES ('compressed-external-direct', repeat('abcdefghijklmnopqrstuvwxyz0123456789', 5000));
+
+SELECT p.descr, pg_column_compression(p.f) AS plain_comp, pg_column_toast_chunk_id(p.f) IS NOT NULL AS plain_has_chunk_id
+FROM tab_intro_plain p
+ORDER BY p.descr;
+
+SELECT d.descr, pg_column_compression(d.f) AS direct_comp, pg_column_toast_chunk_id(d.f) AS direct_chunk_id
+FROM tab_intro_direct d
+ORDER BY d.descr;
+
+DROP TABLE tab_intro_plain;
+DROP TABLE tab_intro_direct;
+
+--
+-- Test Partitioned Tables with Mixed Toast Flavours and Cross-Partition Updates
+--
+CREATE TABLE part_toast(id int, val text) PARTITION BY RANGE (id);
+CREATE TABLE part_toast_p1 PARTITION OF part_toast FOR VALUES FROM (1) TO (100) WITH (toast_flavour = 'direct');
+CREATE TABLE part_toast_p2 PARTITION OF part_toast FOR VALUES FROM (100) TO (200) WITH (toast_flavour = 'plain');
+
+INSERT INTO part_toast SELECT 1, string_agg(md5(i::text), '') FROM generate_series(1, 200) i;
+INSERT INTO part_toast SELECT 101, string_agg(md5(i::text), '') FROM generate_series(1, 200) i;
+
+SELECT id, length(val), substring(val, 1, 10), pg_column_toast_chunk_id(val) IS NOT NULL AS has_chunk_id
+FROM part_toast
+ORDER BY id;
+
+-- Move row from direct partition to plain partition
+UPDATE part_toast SET id = 102 WHERE id = 1;
+SELECT id, length(val), substring(val, 1, 10), pg_column_toast_chunk_id(val) IS NOT NULL AS has_chunk_id
+FROM part_toast
+ORDER BY id;
+
+-- Move row from plain partition to direct partition
+UPDATE part_toast SET id = 2 WHERE id = 101;
+SELECT id, length(val), substring(val, 1, 10), pg_column_toast_chunk_id(val) IS NOT NULL AS has_chunk_id
+FROM part_toast
+ORDER BY id;
+
+DROP TABLE part_toast;
+
+--
+-- Test Expression / Functional Indexes on Direct TOAST Columns
+--
+CREATE TABLE tab_expr_idx(id int primary key, payload text) WITH (toast_flavour = 'direct');
+CREATE INDEX idx_tab_expr_md5 ON tab_expr_idx (md5(payload));
+CREATE INDEX idx_tab_expr_substr ON tab_expr_idx (substring(payload, 1, 20));
+
+INSERT INTO tab_expr_idx VALUES (1, repeat('expr-index-test-payload-', 500));
+INSERT INTO tab_expr_idx VALUES (2, repeat('other-index-test-payload-', 500));
+
+SET enable_seqscan = off;
+
+SELECT id, length(payload) FROM tab_expr_idx WHERE md5(payload) = md5(repeat('expr-index-test-payload-', 500));
+SELECT id, length(payload) FROM tab_expr_idx WHERE substring(payload, 1, 20) = 'expr-index-test-payl';
+
+RESET enable_seqscan;
+
+--
+-- Test Table Maintenance and Rewrites (VACUUM FULL, CLUSTER, ALTER TYPE, TRUNCATE)
+--
+VACUUM FULL tab_expr_idx;
+SELECT id, length(payload), substring(payload, 1, 24) FROM tab_expr_idx ORDER BY id;
+
+CLUSTER tab_expr_idx USING tab_expr_idx_pkey;
+SELECT id, length(payload), substring(payload, 1, 24) FROM tab_expr_idx ORDER BY id;
+
+ALTER TABLE tab_expr_idx ALTER COLUMN payload TYPE varchar(20000);
+SELECT id, length(payload), substring(payload, 1, 24) FROM tab_expr_idx ORDER BY id;
+
+TRUNCATE tab_expr_idx;
+SELECT count(*) FROM tab_expr_idx;
+
+DROP TABLE tab_expr_idx;
+
+--
+-- Test VACUUM FULL / CLUSTER restrictions on direct TOAST tables
+--
+CREATE TABLE tab_toast_maint(id int, val text) WITH (toast_flavour = 'direct');
+INSERT INTO tab_toast_maint VALUES (1, repeat('maint-test-', 500));
+
+-- VACUUM FULL on the parent table succeeds and rebuilds direct toast safely
+VACUUM FULL tab_toast_maint;
+SELECT id, length(val) FROM tab_toast_maint;
+
+-- CLUSTER on direct TOAST table directly is rejected
+DO $$
+DECLARE
+    toast_relname text;
+    toast_idxname text;
+BEGIN
+    SELECT c2.relname, c3.relname INTO toast_relname, toast_idxname
+    FROM pg_class c1
+    JOIN pg_class c2 ON c1.reltoastrelid = c2.oid
+    JOIN pg_index i ON c2.oid = i.indrelid
+    JOIN pg_class c3 ON i.indexrelid = c3.oid
+    WHERE c1.relname = 'tab_toast_maint';
+
+    -- CLUSTER directly on direct TOAST table should be rejected
+    BEGIN
+        EXECUTE 'CLUSTER pg_toast.' || toast_relname || ' USING ' || toast_idxname;
+        RAISE EXCEPTION 'CLUSTER on direct TOAST table should have failed';
+    EXCEPTION WHEN feature_not_supported THEN
+        RAISE NOTICE 'expected error caught for CLUSTER on direct TOAST table: %', regexp_replace(SQLERRM, 'pg_toast_[0-9]+_index', 'pg_toast_xxx_index');
+    END;
+END$$;
+
+DROP TABLE tab_toast_maint;
-- 
2.55.0.1082.g2b9226bbc0-goog