v11-0002-Zstd-compression-and-decompression-routines-incl.patch

From a72c8e8c62e4d5bb8b9de40faf85a8a3422f87dc Mon Sep 17 00:00:00 2001
From: Nikhil Kumar Veldanda <nikhilkv@amazon.com>
Date: Mon, 14 Apr 2025 21:30:03 +0000
Subject: [PATCH v11 2/7] Zstd compression and decompression routines,
 including new catalog table pg_zstd_dictionaries to store zstd dictionaries.

---
 contrib/amcheck/verify_heapam.c               |   1 +
 doc/src/sgml/catalogs.sgml                    |  55 +++
 src/backend/access/brin/brin_tuple.c          |  11 +-
 src/backend/access/common/detoast.c           |  12 +-
 src/backend/access/common/indextuple.c        |   5 +-
 src/backend/access/common/reloptions.c        |  36 +-
 src/backend/access/common/toast_compression.c | 328 +++++++++++++++++-
 src/backend/access/common/toast_internals.c   |  49 ++-
 src/backend/access/table/toast_helper.c       |  15 +-
 src/backend/catalog/Makefile                  |   3 +-
 src/backend/catalog/aclchk.c                  |   2 +
 src/backend/catalog/catalog.c                 |   4 +
 src/backend/catalog/dependency.c              |   2 +
 src/backend/catalog/meson.build               |   1 +
 src/backend/catalog/objectaddress.c           |  76 ++++
 src/backend/catalog/pg_zstd_dictionaries.c    |  56 +++
 src/backend/commands/dropcmds.c               |   1 +
 src/backend/commands/event_trigger.c          |   2 +
 src/backend/commands/seclabel.c               |   1 +
 src/backend/utils/adt/varlena.c               |   3 +
 src/backend/utils/misc/guc_tables.c           |   3 +
 src/backend/utils/misc/postgresql.conf.sample |   2 +-
 src/bin/psql/describe.c                       |   5 +-
 src/bin/psql/tab-complete.in.c                |   4 +-
 src/include/access/toast_compression.h        |  26 +-
 src/include/access/toast_helper.h             |   2 +
 src/include/access/toast_internals.h          |  44 ++-
 src/include/catalog/Makefile                  |   3 +-
 src/include/catalog/meson.build               |   1 +
 src/include/catalog/pg_zstd_dictionaries.h    |  48 +++
 src/include/nodes/parsenodes.h                |   1 +
 src/include/utils/attoptcache.h               |   6 +
 src/test/regress/expected/compression.out     |   5 +-
 src/test/regress/expected/compression_1.out   |   3 +
 src/test/regress/sql/compression.sql          |   1 +
 src/tools/pgindent/typedefs.list              |   2 +
 36 files changed, 774 insertions(+), 45 deletions(-)
 create mode 100644 src/backend/catalog/pg_zstd_dictionaries.c
 create mode 100644 src/include/catalog/pg_zstd_dictionaries.h

diff --git a/contrib/amcheck/verify_heapam.c b/contrib/amcheck/verify_heapam.c
index 2152d8ee577..74991373da9 100644
--- a/contrib/amcheck/verify_heapam.c
+++ b/contrib/amcheck/verify_heapam.c
@@ -1792,6 +1792,7 @@ check_tuple_attribute(HeapCheckContext *ctx)
 				/* List of all valid compression method IDs */
 			case TOAST_PGLZ_COMPRESSION_ID:
 			case TOAST_LZ4_COMPRESSION_ID:
+			case TOAST_ZSTD_COMPRESSION_ID:
 				valid = true;
 				break;
 
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index cbd4e40a320..ab44031d726 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -369,6 +369,12 @@
       <entry><link linkend="catalog-pg-user-mapping"><structname>pg_user_mapping</structname></link></entry>
       <entry>mappings of users to foreign servers</entry>
      </row>
+
+     <row>
+      <entry><link linkend="catalog-pg-zstd-dictionaries"><structname>pg_zstd_dictionaries</structname></link></entry>
+      <entry>Zstandard dictionaries</entry>
+     </row>
+
     </tbody>
    </tgroup>
   </table>
@@ -9779,4 +9785,53 @@ SCRAM-SHA-256$<replaceable>&lt;iteration count&gt;</replaceable>:<replaceable>&l
   </table>
  </sect1>
 
+
+<sect1 id="catalog-pg-zstd-dictionaries">
+  <title><structname>pg_zstd_dictionaries</structname></title>
+
+  <indexterm zone="catalog-pg-zstd-dictionaries">
+    <primary>pg_zstd_dictionaries</primary>
+  </indexterm>
+
+  <para>
+    The catalog <structname>pg_zstd_dictionaries</structname> maintains the dictionaries essential for Zstandard compression and decompression.
+  </para>
+
+  <table>
+    <title><structname>pg_zstd_dictionaries</structname> Columns</title>
+    <tgroup cols="1">
+      <thead>
+        <row>
+          <entry role="catalog_table_entry">
+            <para role="column_definition">Column Type</para>
+            <para>Description</para>
+          </entry>
+        </row>
+      </thead>
+      <tbody>
+        <row>
+          <entry role="catalog_table_entry">
+            <para role="column_definition">
+              <structfield>dictid</structfield> <type>oid</type>
+            </para>
+            <para>
+              Dictionary identifier; a non-null OID that uniquely identifies a dictionary.
+            </para>
+          </entry>
+        </row>
+        <row>
+          <entry role="catalog_table_entry">
+            <para role="column_definition">
+              <structfield>dict</structfield> <type>bytea</type>
+            </para>
+            <para>
+              Variable-length field containing the zstd dictionary data. This field must not be null.
+            </para>
+          </entry>
+        </row>
+      </tbody>
+    </tgroup>
+  </table>
+</sect1>
+
 </chapter>
diff --git a/src/backend/access/brin/brin_tuple.c b/src/backend/access/brin/brin_tuple.c
index 861f397e6db..56099acaa89 100644
--- a/src/backend/access/brin/brin_tuple.c
+++ b/src/backend/access/brin/brin_tuple.c
@@ -222,10 +222,13 @@ brin_form_tuple(BrinDesc *brdesc, BlockNumber blkno, BrinMemTuple *tuple,
 				 atttype->typstorage == TYPSTORAGE_MAIN))
 			{
 				Datum		cvalue;
-				char		compression;
+				CompressionInfo cmp;
+
 				Form_pg_attribute att = TupleDescAttr(brdesc->bd_tupdesc,
 													  keyno);
 
+				setup_compression_info(&cmp, att);
+
 				/*
 				 * If the BRIN summary and indexed attribute use the same data
 				 * type and it has a valid compression method, we can use the
@@ -233,11 +236,11 @@ brin_form_tuple(BrinDesc *brdesc, BlockNumber blkno, BrinMemTuple *tuple,
 				 * default method.
 				 */
 				if (att->atttypid == atttype->type_id)
-					compression = att->attcompression;
+					cmp.cmethod = att->attcompression;
 				else
-					compression = InvalidCompressionMethod;
+					cmp.cmethod = InvalidCompressionMethod;
 
-				cvalue = toast_compress_datum(value, compression);
+				cvalue = toast_compress_datum(value, cmp);
 
 				if (DatumGetPointer(cvalue) != NULL)
 				{
diff --git a/src/backend/access/common/detoast.c b/src/backend/access/common/detoast.c
index 62651787742..b57a9f024c7 100644
--- a/src/backend/access/common/detoast.c
+++ b/src/backend/access/common/detoast.c
@@ -246,10 +246,10 @@ detoast_attr_slice(struct varlena *attr,
 			 * Determine maximum amount of compressed data needed for a prefix
 			 * of a given length (after decompression).
 			 *
-			 * At least for now, if it's LZ4 data, we'll have to fetch the
-			 * whole thing, because there doesn't seem to be an API call to
-			 * determine how much compressed data we need to be sure of being
-			 * able to decompress the required slice.
+			 * At least for now, if it's LZ4 or Zstandard data, we'll have to
+			 * fetch the whole thing, because there doesn't seem to be an API
+			 * call to determine how much compressed data we need to be sure
+			 * of being able to decompress the required slice.
 			 */
 			if (VARATT_EXTERNAL_GET_COMPRESS_METHOD(toast_pointer) ==
 				TOAST_PGLZ_COMPRESSION_ID)
@@ -485,6 +485,8 @@ toast_decompress_datum(struct varlena *attr)
 			return pglz_decompress_datum(attr);
 		case TOAST_LZ4_COMPRESSION_ID:
 			return lz4_decompress_datum(attr);
+		case TOAST_ZSTD_COMPRESSION_ID:
+			return zstd_decompress_datum(attr);
 		default:
 			elog(ERROR, "invalid compression method id %d", cmid);
 			return NULL;		/* keep compiler quiet */
@@ -528,6 +530,8 @@ toast_decompress_datum_slice(struct varlena *attr, int32 slicelength)
 			return pglz_decompress_datum_slice(attr, slicelength);
 		case TOAST_LZ4_COMPRESSION_ID:
 			return lz4_decompress_datum_slice(attr, slicelength);
+		case TOAST_ZSTD_COMPRESSION_ID:
+			return zstd_decompress_datum_slice(attr, slicelength);
 		default:
 			elog(ERROR, "invalid compression method id %d", cmid);
 			return NULL;		/* keep compiler quiet */
diff --git a/src/backend/access/common/indextuple.c b/src/backend/access/common/indextuple.c
index 1986b943a28..cbb57d17799 100644
--- a/src/backend/access/common/indextuple.c
+++ b/src/backend/access/common/indextuple.c
@@ -123,9 +123,10 @@ index_form_tuple_context(TupleDesc tupleDescriptor,
 			 att->attstorage == TYPSTORAGE_MAIN))
 		{
 			Datum		cvalue;
+			CompressionInfo cmp;
 
-			cvalue = toast_compress_datum(untoasted_values[i],
-										  att->attcompression);
+			setup_compression_info(&cmp, att);
+			cvalue = toast_compress_datum(untoasted_values[i], cmp);
 
 			if (DatumGetPointer(cvalue) != NULL)
 			{
diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index 46c1dce222d..345e3dcdb2c 100644
--- a/src/backend/access/common/reloptions.c
+++ b/src/backend/access/common/reloptions.c
@@ -34,6 +34,7 @@
 #include "utils/guc.h"
 #include "utils/memutils.h"
 #include "utils/rel.h"
+#include "access/toast_compression.h"
 
 /*
  * Contents of pg_class.reloptions
@@ -381,7 +382,26 @@ static relopt_int intRelOpts[] =
 		},
 		-1, 0, 1024
 	},
-
+	{
+		{
+			"zstd_dict_size",
+			"Max dict size for zstd",
+			RELOPT_KIND_ATTRIBUTE,
+			ShareUpdateExclusiveLock
+		},
+		DEFAULT_ZSTD_DICT_SIZE, 0, 112640	/* Max dict size(110 KB), 0
+											 * indicates don't use dictionary
+											 * for compression */
+	},
+	{
+		{
+			"zstd_level",
+			"Set column's ZSTD compression level",
+			RELOPT_KIND_ATTRIBUTE,
+			ShareUpdateExclusiveLock
+		},
+		DEFAULT_ZSTD_LEVEL, MIN_ZSTD_LEVEL, MAX_ZSTD_LEVEL
+	},
 	/* list terminator */
 	{{NULL}}
 };
@@ -470,6 +490,15 @@ static relopt_real realRelOpts[] =
 		},
 		0, -1.0, DBL_MAX
 	},
+	{
+		{
+			"dictid",
+			"Current dictid for column",
+			RELOPT_KIND_ATTRIBUTE,
+			ShareUpdateExclusiveLock
+		},
+		InvalidDictId, InvalidDictId, UINT32_MAX
+	},
 	{
 		{
 			"vacuum_cleanup_index_scale_factor",
@@ -2097,7 +2126,10 @@ attribute_reloptions(Datum reloptions, bool validate)
 {
 	static const relopt_parse_elt tab[] = {
 		{"n_distinct", RELOPT_TYPE_REAL, offsetof(AttributeOpts, n_distinct)},
-		{"n_distinct_inherited", RELOPT_TYPE_REAL, offsetof(AttributeOpts, n_distinct_inherited)}
+		{"n_distinct_inherited", RELOPT_TYPE_REAL, offsetof(AttributeOpts, n_distinct_inherited)},
+		{"dictid", RELOPT_TYPE_REAL, offsetof(AttributeOpts, dictid)},
+		{"zstd_dict_size", RELOPT_TYPE_INT, offsetof(AttributeOpts, zstd_dict_size)},
+		{"zstd_level", RELOPT_TYPE_INT, offsetof(AttributeOpts, zstd_level)},
 	};
 
 	return (bytea *) build_reloptions(reloptions, validate,
diff --git a/src/backend/access/common/toast_compression.c b/src/backend/access/common/toast_compression.c
index 21f2f4af97e..3b028993f0b 100644
--- a/src/backend/access/common/toast_compression.c
+++ b/src/backend/access/common/toast_compression.c
@@ -17,19 +17,41 @@
 #include <lz4.h>
 #endif
 
+#ifdef USE_ZSTD
+#include <zstd.h>
+#endif
+
 #include "access/detoast.h"
 #include "access/toast_compression.h"
 #include "common/pg_lzcompress.h"
 #include "varatt.h"
+#include "catalog/pg_zstd_dictionaries.h"
+#include "access/toast_internals.h"
 
 /* GUC */
 int			default_toast_compression = TOAST_PGLZ_COMPRESSION;
 
-#define NO_LZ4_SUPPORT() \
+#ifdef USE_ZSTD
+static ZSTD_CCtx *ZstdCompressionCtx = NULL;
+static ZSTD_CDict * ZstdCompressionCtxCDict = NULL;
+static Oid	ZstdCompressionCtxDictID = InvalidDictId;
+
+static ZSTD_DCtx *ZstdDecompressionCtx = NULL;
+static ZSTD_DDict * ZstdDecompressionCtxDDict = NULL;
+static Oid	ZstdDecompressionCtxDictID = InvalidDictId;
+
+#define ZSTD_CHECK_ERROR(zstd_ret, msg) \
+	do { \
+		if (ZSTD_isError(zstd_ret)) \
+			ereport(ERROR, (errmsg("%s: %s", (msg), ZSTD_getErrorName(zstd_ret)))); \
+	} while (0)
+#endif
+
+#define COMPRESSION_METHOD_NOT_SUPPORTED(method) \
 	ereport(ERROR, \
 			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED), \
-			 errmsg("compression method lz4 not supported"), \
-			 errdetail("This functionality requires the server to be built with lz4 support.")))
+			 errmsg("compression method %s not supported", method), \
+			 errdetail("This functionality requires the server to be built with %s support.", method)))
 
 /*
  * Compress a varlena using PGLZ.
@@ -139,7 +161,7 @@ struct varlena *
 lz4_compress_datum(const struct varlena *value)
 {
 #ifndef USE_LZ4
-	NO_LZ4_SUPPORT();
+	COMPRESSION_METHOD_NOT_SUPPORTED("lz4");
 	return NULL;				/* keep compiler quiet */
 #else
 	int32		valsize;
@@ -182,7 +204,7 @@ struct varlena *
 lz4_decompress_datum(const struct varlena *value)
 {
 #ifndef USE_LZ4
-	NO_LZ4_SUPPORT();
+	COMPRESSION_METHOD_NOT_SUPPORTED("lz4");
 	return NULL;				/* keep compiler quiet */
 #else
 	int32		rawsize;
@@ -215,7 +237,7 @@ struct varlena *
 lz4_decompress_datum_slice(const struct varlena *value, int32 slicelength)
 {
 #ifndef USE_LZ4
-	NO_LZ4_SUPPORT();
+	COMPRESSION_METHOD_NOT_SUPPORTED("lz4");
 	return NULL;				/* keep compiler quiet */
 #else
 	int32		rawsize;
@@ -266,7 +288,13 @@ toast_get_compression_id(struct varlena *attr)
 
 		VARATT_EXTERNAL_GET_POINTER(toast_pointer, attr);
 
-		if (VARATT_EXTERNAL_IS_COMPRESSED(toast_pointer))
+		if (VARATT_EXTERNAL_IS_COMPRESSED(toast_pointer) && VARATT_EXTERNAL_GET_COMPRESS_METHOD(toast_pointer) >= TOAST_ZSTD_COMPRESSION_ID)
+		{
+			struct varlena *compressed_attr = detoast_external_attr(attr);
+
+			cmid = TOAST_COMPRESS_METHOD(compressed_attr);
+		}
+		else
 			cmid = VARATT_EXTERNAL_GET_COMPRESS_METHOD(toast_pointer);
 	}
 	else if (VARATT_IS_COMPRESSED(attr))
@@ -289,10 +317,17 @@ CompressionNameToMethod(const char *compression)
 	else if (strcmp(compression, "lz4") == 0)
 	{
 #ifndef USE_LZ4
-		NO_LZ4_SUPPORT();
+		COMPRESSION_METHOD_NOT_SUPPORTED("lz4");
 #endif
 		return TOAST_LZ4_COMPRESSION;
 	}
+	else if (strcmp(compression, "zstd") == 0)
+	{
+#ifndef USE_ZSTD
+		COMPRESSION_METHOD_NOT_SUPPORTED("zstd");
+#endif
+		return TOAST_ZSTD_COMPRESSION;
+	}
 
 	return InvalidCompressionMethod;
 }
@@ -309,8 +344,285 @@ GetCompressionMethodName(char method)
 			return "pglz";
 		case TOAST_LZ4_COMPRESSION:
 			return "lz4";
+		case TOAST_ZSTD_COMPRESSION:
+			return "zstd";
 		default:
 			elog(ERROR, "invalid compression method %c", method);
 			return NULL;		/* keep compiler quiet */
 	}
 }
+
+/* Compress datum using ZSTD with optional dictionary (using cdict) */
+struct varlena *
+zstd_compress_datum(const struct varlena *value, Oid dictid, int zstd_level)
+{
+#ifdef USE_ZSTD
+	uint32		valsize = VARSIZE_ANY_EXHDR(value);
+	size_t		max_size = ZSTD_compressBound(valsize);
+	struct varlena *compressed;
+	void	   *dest;
+	size_t		cmp_size,
+				ret;
+
+	/* Create the session CCtx if it hasn't been yet */
+	if (ZstdCompressionCtx == NULL)
+	{
+		ZstdCompressionCtx = ZSTD_createCCtx();
+		if (!ZstdCompressionCtx)
+			ereport(ERROR, (errmsg("could not create ZSTD_CCtx")));
+		ZstdCompressionCtxDictID = InvalidDictId;
+	}
+
+	/* Reset the context to clear any prior state */
+	ret = ZSTD_CCtx_reset(ZstdCompressionCtx, ZSTD_reset_session_only);
+	ZSTD_CHECK_ERROR(ret, "failed to reset ZSTD CCtx");
+
+	/* Set compression level */
+	ret = ZSTD_CCtx_setParameter(ZstdCompressionCtx, ZSTD_c_compressionLevel, zstd_level);
+	ZSTD_CHECK_ERROR(ret, "failed to set ZSTD compression level");
+
+	/* Check and update dictionary if changed */
+	if (ZstdCompressionCtxDictID != dictid)
+	{
+		/* If there's a previous dictionary, detach and free it */
+		if (ZstdCompressionCtxCDict)
+		{
+			ZSTD_freeCDict(ZstdCompressionCtxCDict);
+			ZstdCompressionCtxCDict = NULL;
+		}
+
+		if (dictid != InvalidDictId)
+		{
+			bytea	   *dict_bytea = get_zstd_dict(dictid);
+			const void *dict_buffer = VARDATA_ANY(dict_bytea);
+			uint32		dict_size = VARSIZE_ANY(dict_bytea) - VARHDRSZ;
+			ZSTD_CDict *cdict = ZSTD_createCDict(dict_buffer, dict_size, zstd_level);
+
+			pfree(dict_bytea);
+
+			if (!cdict)
+				ereport(ERROR, (errmsg("Failed to create ZSTD compression dictionary")));
+
+			ret = ZSTD_CCtx_refCDict(ZstdCompressionCtx, cdict);
+			ZSTD_CHECK_ERROR(ret, "failed to load ZSTD dictionary");
+
+			ZstdCompressionCtxCDict = cdict;
+		}
+		else
+		{
+			/* Unload any previously used dictionary by passing NULL */
+			ret = ZSTD_CCtx_refCDict(ZstdCompressionCtx, NULL);
+			ZSTD_CHECK_ERROR(ret, "failed to unload ZSTD dictionary");
+		}
+
+		ZstdCompressionCtxDictID = dictid;
+	}
+
+	/* Allocate space for the compressed varlena (header + data) */
+	compressed = (struct varlena *) palloc(max_size + VARHDRSZ_COMPRESSED_EXT);
+	dest = (char *) compressed + VARHDRSZ_COMPRESSED_EXT;
+
+	cmp_size = ZSTD_compress2(ZstdCompressionCtx,
+							  dest,
+							  max_size,
+							  VARDATA_ANY(value),
+							  valsize);
+
+	if (ZSTD_isError(cmp_size))
+	{
+		pfree(compressed);
+		ZSTD_CHECK_ERROR(cmp_size, "ZSTD compression failed");
+	}
+
+	/*
+	 * If compression did not reduce size, return NULL so that the
+	 * uncompressed data is stored
+	 */
+	if (cmp_size > valsize)
+	{
+		pfree(compressed);
+		return NULL;
+	}
+
+	/* Set the compressed size in the varlena header */
+	SET_VARSIZE_COMPRESSED(compressed, cmp_size + VARHDRSZ_COMPRESSED_EXT);
+	return compressed;
+
+#else
+	COMPRESSION_METHOD_NOT_SUPPORTED("zstd");
+	return NULL;
+#endif
+}
+
+/* Decompression routine */
+struct varlena *
+zstd_decompress_datum(const struct varlena *value)
+{
+#ifdef USE_ZSTD
+	uint32		actual_size_exhdr = VARDATA_COMPRESSED_GET_EXTSIZE(value);
+	uint32		cmp_size_exhdr = VARSIZE_4B(value) - VARHDRSZ_COMPRESSED_EXT;
+	Oid			dictid = (Oid) VARDATA_COMPRESSED_GET_DICTID(value);
+	struct varlena *result;
+	size_t		uncmp_size,
+				ret;
+
+	if (ZstdDecompressionCtx == NULL)
+	{
+		ZstdDecompressionCtx = ZSTD_createDCtx();
+		if (!ZstdDecompressionCtx)
+			ereport(ERROR, (errmsg("could not create ZSTD_DCtx")));
+		ZstdDecompressionCtxDictID = InvalidDictId;
+	}
+
+	/* Reset the context to clear any prior state */
+	ret = ZSTD_DCtx_reset(ZstdDecompressionCtx, ZSTD_reset_session_only);
+	ZSTD_CHECK_ERROR(ret, "failed to reset ZSTD DCtx");
+
+	if (ZstdDecompressionCtxDictID != dictid)
+	{
+		if (ZstdDecompressionCtxDDict)
+		{
+			ZSTD_freeDDict(ZstdDecompressionCtxDDict);
+			ZstdDecompressionCtxDDict = NULL;
+		}
+
+		if (dictid != InvalidDictId)
+		{
+			bytea	   *dict_bytea = get_zstd_dict(dictid);
+			const void *dict_buffer = VARDATA_ANY(dict_bytea);
+			uint32		dict_size = VARSIZE_ANY(dict_bytea) - VARHDRSZ;
+			ZSTD_DDict *ddict = ZSTD_createDDict(dict_buffer, dict_size);
+
+			pfree(dict_bytea);
+
+			if (!ddict)
+				ereport(ERROR, (errmsg("Failed to create ZSTD decompression dictionary")));
+
+			ret = ZSTD_DCtx_refDDict(ZstdDecompressionCtx, ddict);
+			ZSTD_CHECK_ERROR(ret, "failed to load ZSTD dictionary");
+
+			ZstdDecompressionCtxDDict = ddict;
+		}
+		else
+		{
+			/* Unload any previously used dictionary by passing NULL */
+			ret = ZSTD_DCtx_refDDict(ZstdDecompressionCtx, NULL);
+			ZSTD_CHECK_ERROR(ret, "failed to unload ZSTD dictionary");
+		}
+
+		/* Update the tracked dictionary ID */
+		ZstdDecompressionCtxDictID = dictid;
+	}
+
+	/* Allocate space for the uncompressed data */
+	result = (struct varlena *) palloc(actual_size_exhdr + VARHDRSZ);
+
+	uncmp_size = ZSTD_decompressDCtx(ZstdDecompressionCtx,
+									 VARDATA(result),
+									 actual_size_exhdr,
+									 VARDATA_4B_C(value),
+									 cmp_size_exhdr);
+
+	if (ZSTD_isError(uncmp_size))
+	{
+		pfree(result);
+		ZSTD_CHECK_ERROR(uncmp_size, "ZSTD decompression failed");
+	}
+
+	/* Set final size in the varlena header */
+	SET_VARSIZE(result, uncmp_size + VARHDRSZ);
+	return result;
+
+#else
+	COMPRESSION_METHOD_NOT_SUPPORTED("zstd");
+	return NULL;
+#endif
+}
+
+/* Decompress a slice of the datum using the streaming API and optional dictionary */
+struct varlena *
+zstd_decompress_datum_slice(const struct varlena *value, int32 slicelength)
+{
+#ifdef USE_ZSTD
+	struct varlena *result;
+	ZSTD_inBuffer inBuf;
+	ZSTD_outBuffer outBuf;
+	Oid			dictid = (Oid) VARDATA_COMPRESSED_GET_DICTID(value);
+	size_t		ret;
+
+	if (ZstdDecompressionCtx == NULL)
+	{
+		ZstdDecompressionCtx = ZSTD_createDCtx();
+		if (!ZstdDecompressionCtx)
+			elog(ERROR, "could not create ZSTD_DCtx");
+		ZstdDecompressionCtxDictID = InvalidDictId;
+	}
+
+	/* Reset the context to clear any prior state */
+	ret = ZSTD_DCtx_reset(ZstdDecompressionCtx, ZSTD_reset_session_only);
+	ZSTD_CHECK_ERROR(ret, "failed to reset ZSTD_DCtx");
+
+	if (ZstdDecompressionCtxDictID != dictid)
+	{
+		if (ZstdDecompressionCtxDDict)
+		{
+			ZSTD_freeDDict(ZstdDecompressionCtxDDict);
+			ZstdDecompressionCtxDDict = NULL;
+		}
+
+		if (dictid != InvalidDictId)
+		{
+			bytea	   *dict_bytea = get_zstd_dict(dictid);
+			const void *dict_buffer = VARDATA_ANY(dict_bytea);
+			uint32		dict_size = VARSIZE_ANY(dict_bytea) - VARHDRSZ;
+			ZSTD_DDict *ddict = ZSTD_createDDict(dict_buffer, dict_size);
+
+			pfree(dict_bytea);
+
+			if (!ddict)
+				ereport(ERROR, (errmsg("Failed to create ZSTD decompression dictionary")));
+
+			ret = ZSTD_DCtx_refDDict(ZstdDecompressionCtx, ddict);
+			ZSTD_CHECK_ERROR(ret, "failed to load ZSTD dictionary");
+
+			ZstdDecompressionCtxDDict = ddict;
+		}
+		else
+		{
+			/* Unload any previously used dictionary by passing NULL */
+			ret = ZSTD_DCtx_refDDict(ZstdDecompressionCtx, NULL);
+			ZSTD_CHECK_ERROR(ret, "failed to unload ZSTD dictionary");
+		}
+
+		/* Update the tracked dictionary ID */
+		ZstdDecompressionCtxDictID = dictid;
+	}
+
+	inBuf.src = (char *) value + VARHDRSZ_COMPRESSED_EXT;
+	inBuf.size = VARSIZE(value) - VARHDRSZ_COMPRESSED_EXT;
+	inBuf.pos = 0;
+
+	result = (struct varlena *) palloc(slicelength + VARHDRSZ);
+	outBuf.dst = (char *) result + VARHDRSZ;
+	outBuf.size = slicelength;
+	outBuf.pos = 0;
+
+	/* Common decompression loop */
+	while (inBuf.pos < inBuf.size && outBuf.pos < outBuf.size)
+	{
+		ret = ZSTD_decompressStream(ZstdDecompressionCtx, &outBuf, &inBuf);
+		if (ZSTD_isError(ret))
+		{
+			pfree(result);
+			ZSTD_CHECK_ERROR(ret, "zstd decompression failed");
+		}
+	}
+
+	Assert(outBuf.size == slicelength && outBuf.pos == slicelength);
+	SET_VARSIZE(result, outBuf.pos + VARHDRSZ);
+	return result;
+#else
+	COMPRESSION_METHOD_NOT_SUPPORTED("zstd");
+	return NULL;
+#endif
+}
diff --git a/src/backend/access/common/toast_internals.c b/src/backend/access/common/toast_internals.c
index 7d8be8346ce..46f0381b2de 100644
--- a/src/backend/access/common/toast_internals.c
+++ b/src/backend/access/common/toast_internals.c
@@ -25,6 +25,7 @@
 #include "utils/fmgroids.h"
 #include "utils/rel.h"
 #include "utils/snapmgr.h"
+#include "utils/attoptcache.h"
 
 static bool toastrel_valueid_exists(Relation toastrel, Oid valueid);
 static bool toastid_valueid_exists(Oid toastrelid, Oid valueid);
@@ -43,7 +44,7 @@ static bool toastid_valueid_exists(Oid toastrelid, Oid valueid);
  * ----------
  */
 Datum
-toast_compress_datum(Datum value, char cmethod)
+toast_compress_datum(Datum value, CompressionInfo cmp)
 {
 	struct varlena *tmp = NULL;
 	int32		valsize;
@@ -55,13 +56,13 @@ toast_compress_datum(Datum value, char cmethod)
 	valsize = VARSIZE_ANY_EXHDR(DatumGetPointer(value));
 
 	/* If the compression method is not valid, use the current default */
-	if (!CompressionMethodIsValid(cmethod))
-		cmethod = default_toast_compression;
+	if (!CompressionMethodIsValid(cmp.cmethod))
+		cmp.cmethod = default_toast_compression;
 
 	/*
 	 * Call appropriate compression routine for the compression method.
 	 */
-	switch (cmethod)
+	switch (cmp.cmethod)
 	{
 		case TOAST_PGLZ_COMPRESSION:
 			tmp = pglz_compress_datum((const struct varlena *) value);
@@ -71,8 +72,12 @@ toast_compress_datum(Datum value, char cmethod)
 			tmp = lz4_compress_datum((const struct varlena *) value);
 			cmid = TOAST_LZ4_COMPRESSION_ID;
 			break;
+		case TOAST_ZSTD_COMPRESSION:
+			tmp = zstd_compress_datum((const struct varlena *) value, cmp.dictid, cmp.zstd_level);
+			cmid = TOAST_ZSTD_COMPRESSION_ID;
+			break;
 		default:
-			elog(ERROR, "invalid compression method %c", cmethod);
+			elog(ERROR, "invalid compression method %c", cmp.cmethod);
 	}
 
 	if (tmp == NULL)
@@ -90,9 +95,11 @@ toast_compress_datum(Datum value, char cmethod)
 	 */
 	if (VARSIZE(tmp) < valsize - 2)
 	{
+		Oid			dictid = cmp.dictid;
+
 		/* successful compression */
 		Assert(cmid != TOAST_INVALID_COMPRESSION_ID);
-		TOAST_COMPRESS_SET_SIZE_AND_COMPRESS_METHOD(tmp, valsize, cmid);
+		TOAST_COMPRESS_SET_SIZE_AND_COMPRESS_METHOD(tmp, valsize, cmid, dictid);
 		return PointerGetDatum(tmp);
 	}
 	else
@@ -654,3 +661,33 @@ get_toast_snapshot(void)
 
 	return &SnapshotToastData;
 }
+
+void
+setup_compression_info(CompressionInfo * info, Form_pg_attribute att)
+{
+	info->cmethod = att->attcompression;
+	info->dictid = InvalidDictId;
+	info->zstd_level = DEFAULT_ZSTD_LEVEL;
+
+	if (att->attcompression == TOAST_ZSTD_COMPRESSION)
+	{
+		AttributeOpts *aopt = get_attribute_options(att->attrelid, att->attnum);
+
+		if (aopt != NULL)
+		{
+			info->zstd_level = aopt->zstd_level;
+			/**
+			 * If user marks zstd dict size as 0, then we don't use dict compression
+			 * for this attribute.
+			 */
+			if (aopt->zstd_dict_size != 0)
+			{
+				if (aopt->dictid < InvalidDictId || aopt->dictid > UINT32_MAX)
+					ereport(ERROR,
+							(errcode(ERRCODE_INTERNAL_ERROR),
+							 errmsg("dictid is not in expected range")));
+				info->dictid = (Oid) aopt->dictid;
+			}
+		}
+	}
+}
diff --git a/src/backend/access/table/toast_helper.c b/src/backend/access/table/toast_helper.c
index b60fab0a4d2..f4b1cbe494e 100644
--- a/src/backend/access/table/toast_helper.c
+++ b/src/backend/access/table/toast_helper.c
@@ -51,10 +51,15 @@ toast_tuple_init(ToastTupleContext *ttc)
 		Form_pg_attribute att = TupleDescAttr(tupleDesc, i);
 		struct varlena *old_value;
 		struct varlena *new_value;
+		CompressionInfo cmp;
+
+		setup_compression_info(&cmp, att);
 
 		ttc->ttc_attr[i].tai_colflags = 0;
 		ttc->ttc_attr[i].tai_oldexternal = NULL;
-		ttc->ttc_attr[i].tai_compression = att->attcompression;
+		ttc->ttc_attr[i].tai_compression = cmp.cmethod;
+		ttc->ttc_attr[i].dictid = cmp.dictid;
+		ttc->ttc_attr[i].zstd_level = cmp.zstd_level;
 
 		if (ttc->ttc_oldvalues != NULL)
 		{
@@ -230,7 +235,13 @@ toast_tuple_try_compression(ToastTupleContext *ttc, int attribute)
 	Datum		new_value;
 	ToastAttrInfo *attr = &ttc->ttc_attr[attribute];
 
-	new_value = toast_compress_datum(*value, attr->tai_compression);
+	CompressionInfo cmp = {
+		.cmethod = attr->tai_compression,
+		.dictid = attr->dictid,
+		.zstd_level = attr->zstd_level
+	};
+
+	new_value = toast_compress_datum(*value, cmp);
 
 	if (DatumGetPointer(new_value) != NULL)
 	{
diff --git a/src/backend/catalog/Makefile b/src/backend/catalog/Makefile
index c090094ed08..282afbcef5e 100644
--- a/src/backend/catalog/Makefile
+++ b/src/backend/catalog/Makefile
@@ -46,7 +46,8 @@ OBJS = \
 	pg_subscription.o \
 	pg_type.o \
 	storage.o \
-	toasting.o
+	toasting.o \
+	pg_zstd_dictionaries.o
 
 include $(top_srcdir)/src/backend/common.mk
 
diff --git a/src/backend/catalog/aclchk.c b/src/backend/catalog/aclchk.c
index 9ca8a88dc91..01d70ebc53e 100644
--- a/src/backend/catalog/aclchk.c
+++ b/src/backend/catalog/aclchk.c
@@ -2771,6 +2771,7 @@ aclcheck_error(AclResult aclerr, ObjectType objtype,
 					case OBJECT_TSPARSER:
 					case OBJECT_TSTEMPLATE:
 					case OBJECT_USER_MAPPING:
+					case OBJECT_ZSTD_DICTIONARY:
 						elog(ERROR, "unsupported object type: %d", objtype);
 				}
 
@@ -2909,6 +2910,7 @@ aclcheck_error(AclResult aclerr, ObjectType objtype,
 					case OBJECT_TSPARSER:
 					case OBJECT_TSTEMPLATE:
 					case OBJECT_USER_MAPPING:
+					case OBJECT_ZSTD_DICTIONARY:
 						elog(ERROR, "unsupported object type: %d", objtype);
 				}
 
diff --git a/src/backend/catalog/catalog.c b/src/backend/catalog/catalog.c
index a6edf614606..d89334f7e87 100644
--- a/src/backend/catalog/catalog.c
+++ b/src/backend/catalog/catalog.c
@@ -40,6 +40,7 @@
 #include "catalog/pg_subscription.h"
 #include "catalog/pg_tablespace.h"
 #include "catalog/pg_type.h"
+#include "catalog/pg_zstd_dictionaries_d.h"
 #include "miscadmin.h"
 #include "utils/fmgroids.h"
 #include "utils/fmgrprotos.h"
@@ -381,6 +382,9 @@ IsPinnedObject(Oid classId, Oid objectId)
 	if (classId == DatabaseRelationId)
 		return false;
 
+	if (classId == ZstdDictionariesRelationId)
+		return false;
+
 	/*
 	 * All other initdb-created objects are pinned.  This is overkill (the
 	 * system doesn't really depend on having every last weird datatype, for
diff --git a/src/backend/catalog/dependency.c b/src/backend/catalog/dependency.c
index 18316a3968b..0ea61ed1dae 100644
--- a/src/backend/catalog/dependency.c
+++ b/src/backend/catalog/dependency.c
@@ -65,6 +65,7 @@
 #include "catalog/pg_ts_template.h"
 #include "catalog/pg_type.h"
 #include "catalog/pg_user_mapping.h"
+#include "catalog/pg_zstd_dictionaries_d.h"
 #include "commands/comment.h"
 #include "commands/defrem.h"
 #include "commands/event_trigger.h"
@@ -1464,6 +1465,7 @@ doDeletion(const ObjectAddress *object, int flags)
 		case EventTriggerRelationId:
 		case TransformRelationId:
 		case AuthMemRelationId:
+		case ZstdDictionariesRelationId:
 			DropObjectById(object);
 			break;
 
diff --git a/src/backend/catalog/meson.build b/src/backend/catalog/meson.build
index 1958ea9238a..8f0413189cb 100644
--- a/src/backend/catalog/meson.build
+++ b/src/backend/catalog/meson.build
@@ -34,6 +34,7 @@ backend_sources += files(
   'pg_type.c',
   'storage.c',
   'toasting.c',
+  'pg_zstd_dictionaries.c',
 )
 
 
diff --git a/src/backend/catalog/objectaddress.c b/src/backend/catalog/objectaddress.c
index b63fd57dc04..ba3e52e5323 100644
--- a/src/backend/catalog/objectaddress.c
+++ b/src/backend/catalog/objectaddress.c
@@ -62,6 +62,7 @@
 #include "catalog/pg_ts_template.h"
 #include "catalog/pg_type.h"
 #include "catalog/pg_user_mapping.h"
+#include "catalog/pg_zstd_dictionaries.h"
 #include "commands/dbcommands.h"
 #include "commands/defrem.h"
 #include "commands/event_trigger.h"
@@ -636,6 +637,20 @@ static const ObjectPropertyType ObjectProperty[] =
 		OBJECT_USER_MAPPING,
 		false
 	},
+	{
+		"zstd dictionaries",
+		ZstdDictionariesRelationId,
+		ZstdDictidIndexId,
+		ZSTDDICTIDOID,
+		-1,
+		Anum_pg_zstd_dictionaries_dictid,
+		InvalidAttrNumber,
+		InvalidAttrNumber,
+		InvalidAttrNumber,
+		InvalidAttrNumber,
+		OBJECT_ZSTD_DICTIONARY,
+		true
+	},
 };
 
 /*
@@ -831,6 +846,9 @@ static const struct object_type_map
 	},
 	{
 		"statistics object", OBJECT_STATISTIC_EXT
+	},
+	{
+		"zstd dictionary", OBJECT_ZSTD_DICTIONARY
 	}
 };
 
@@ -1127,6 +1145,26 @@ get_object_address(ObjectType objtype, Node *object,
 															 missing_ok);
 				address.objectSubId = 0;
 				break;
+			case OBJECT_ZSTD_DICTIONARY:
+				{
+					Oid			dictid = oidparse(object);
+					HeapTuple	tuple = SearchSysCache1(ZSTDDICTIDOID, ObjectIdGetDatum(dictid));
+
+					if (!HeapTupleIsValid(tuple))
+					{
+						if (!missing_ok)
+							ereport(ERROR,
+									(errcode(ERRCODE_UNDEFINED_OBJECT),
+									 errmsg("zstd dictionary %u does not exist",
+											dictid)));
+					}
+					ReleaseSysCache(tuple);
+
+					address.classId = ZstdDictionariesRelationId;
+					address.objectId = dictid;
+					address.objectSubId = 0;
+					break;
+				}
 				/* no default, to let compiler warn about missing case */
 		}
 
@@ -2172,6 +2210,23 @@ pg_get_object_address(PG_FUNCTION_ARGS)
 					 errmsg("large object OID may not be null")));
 		objnode = (Node *) makeFloat(TextDatumGetCString(elems[0]));
 	}
+	else if (type == OBJECT_ZSTD_DICTIONARY)
+	{
+		Datum	   *elems;
+		bool	   *nulls;
+		int			nelems;
+
+		deconstruct_array_builtin(namearr, TEXTOID, &elems, &nulls, &nelems);
+		if (nelems != 1)
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("name list length must be exactly %d", 1)));
+		if (nulls[0])
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("zstd dictionary OID may not be null")));
+		objnode = (Node *) makeFloat(TextDatumGetCString(elems[0]));
+	}
 	else
 	{
 		name = textarray_to_strvaluelist(namearr);
@@ -2354,6 +2409,7 @@ pg_get_object_address(PG_FUNCTION_ARGS)
 				break;
 			}
 		case OBJECT_LARGEOBJECT:
+		case OBJECT_ZSTD_DICTIONARY:
 			/* already handled above */
 			break;
 			/* no default, to let compiler warn about missing case */
@@ -2557,6 +2613,7 @@ check_object_ownership(Oid roleid, ObjectType objtype, ObjectAddress address,
 		case OBJECT_PUBLICATION_NAMESPACE:
 		case OBJECT_PUBLICATION_REL:
 		case OBJECT_USER_MAPPING:
+		case OBJECT_ZSTD_DICTIONARY:
 			/* These are currently not supported or don't make sense here. */
 			elog(ERROR, "unsupported object type: %d", (int) objtype);
 			break;
@@ -4067,7 +4124,26 @@ getObjectDescription(const ObjectAddress *object, bool missing_ok)
 				ReleaseSysCache(trfTup);
 				break;
 			}
+		case ZstdDictionariesRelationId:
+			{
+				HeapTuple	htup;
+				Form_pg_zstd_dictionaries zstd;
 
+				htup = SearchSysCache1(ZSTDDICTIDOID, ObjectIdGetDatum(object->objectId));
+				if (!HeapTupleIsValid(htup))
+				{
+					if (!missing_ok)
+						elog(ERROR, "could not find tuple for dictid %u",
+							 object->objectId);
+					break;
+				}
+
+				zstd = (Form_pg_zstd_dictionaries) GETSTRUCT(htup);
+				appendStringInfo(&buffer, _("Dictionary Id %d"), zstd->dictid);
+
+				ReleaseSysCache(htup);
+				break;
+			}
 		default:
 			elog(ERROR, "unsupported object class: %u", object->classId);
 	}
diff --git a/src/backend/catalog/pg_zstd_dictionaries.c b/src/backend/catalog/pg_zstd_dictionaries.c
new file mode 100644
index 00000000000..d5e965c34d0
--- /dev/null
+++ b/src/backend/catalog/pg_zstd_dictionaries.c
@@ -0,0 +1,56 @@
+/*-------------------------------------------------------------------------
+ *
+ * pg_zstd_dictionaries.c
+ *	  routines to support manipulation of the pg_zstd_dictionaries relation
+ *
+ * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/catalog/pg_zstd_dictionaries.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "fmgr.h"
+#include "catalog/pg_zstd_dictionaries.h"
+#include "catalog/pg_zstd_dictionaries_d.h"
+#include "utils/syscache.h"
+#include "varatt.h"
+
+/*
+ * get_zstd_dict - Fetches the ZSTD dictionary from the catalog
+ *
+ * dictid: The Oid of the dictionary to fetch.
+ *
+ * Returns: A pointer to a bytea containing the dictionary data.
+ */
+bytea *
+get_zstd_dict(Oid dictid)
+{
+	HeapTuple	tuple;
+	Datum		datum;
+	bool		isNull;
+	bytea	   *dict_bytea;
+	Size		bytea_len;
+	bytea	   *result;
+
+	tuple = SearchSysCache1(ZSTDDICTIDOID, ObjectIdGetDatum(dictid));
+	if (!HeapTupleIsValid(tuple))
+		ereport(ERROR, (errmsg("Cache lookup failed for dictid %u", dictid)));
+
+	datum = SysCacheGetAttr(ZSTDDICTIDOID, tuple, Anum_pg_zstd_dictionaries_dict, &isNull);
+	if (isNull)
+		ereport(ERROR, (errmsg("Dictionary not found for dictid %u", dictid)));
+
+	dict_bytea = DatumGetByteaP(datum);
+	bytea_len = VARSIZE(dict_bytea);
+
+	result = palloc(bytea_len);
+	memcpy(result, dict_bytea, bytea_len);
+
+	ReleaseSysCache(tuple);
+
+	return result;
+}
diff --git a/src/backend/commands/dropcmds.c b/src/backend/commands/dropcmds.c
index ceb9a229b63..58bc92d5959 100644
--- a/src/backend/commands/dropcmds.c
+++ b/src/backend/commands/dropcmds.c
@@ -508,6 +508,7 @@ does_not_exist_skipping(ObjectType objtype, Node *object)
 		case OBJECT_PUBLICATION_REL:
 		case OBJECT_TABCONSTRAINT:
 		case OBJECT_USER_MAPPING:
+		case OBJECT_ZSTD_DICTIONARY:
 			/* These are currently not used or needed. */
 			elog(ERROR, "unsupported object type: %d", (int) objtype);
 			break;
diff --git a/src/backend/commands/event_trigger.c b/src/backend/commands/event_trigger.c
index edc2c988e29..c5c9291c384 100644
--- a/src/backend/commands/event_trigger.c
+++ b/src/backend/commands/event_trigger.c
@@ -2186,6 +2186,7 @@ stringify_grant_objtype(ObjectType objtype)
 		case OBJECT_TSTEMPLATE:
 		case OBJECT_USER_MAPPING:
 		case OBJECT_VIEW:
+		case OBJECT_ZSTD_DICTIONARY:
 			elog(ERROR, "unsupported object type: %d", (int) objtype);
 	}
 
@@ -2270,6 +2271,7 @@ stringify_adefprivs_objtype(ObjectType objtype)
 		case OBJECT_TSTEMPLATE:
 		case OBJECT_USER_MAPPING:
 		case OBJECT_VIEW:
+		case OBJECT_ZSTD_DICTIONARY:
 			elog(ERROR, "unsupported object type: %d", (int) objtype);
 	}
 
diff --git a/src/backend/commands/seclabel.c b/src/backend/commands/seclabel.c
index cee5d7bbb9c..e4fe6a64b2e 100644
--- a/src/backend/commands/seclabel.c
+++ b/src/backend/commands/seclabel.c
@@ -92,6 +92,7 @@ SecLabelSupportsObjectType(ObjectType objtype)
 		case OBJECT_TSPARSER:
 		case OBJECT_TSTEMPLATE:
 		case OBJECT_USER_MAPPING:
+		case OBJECT_ZSTD_DICTIONARY:
 			return false;
 
 			/*
diff --git a/src/backend/utils/adt/varlena.c b/src/backend/utils/adt/varlena.c
index 3e4d5568bde..063780e56dc 100644
--- a/src/backend/utils/adt/varlena.c
+++ b/src/backend/utils/adt/varlena.c
@@ -5301,6 +5301,9 @@ pg_column_compression(PG_FUNCTION_ARGS)
 		case TOAST_LZ4_COMPRESSION_ID:
 			result = "lz4";
 			break;
+		case TOAST_ZSTD_COMPRESSION_ID:
+			result = "zstd";
+			break;
 		default:
 			elog(ERROR, "invalid compression method id %d", cmid);
 	}
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index 60b12446a1c..b4b0e44d7ea 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -460,6 +460,9 @@ static const struct config_enum_entry default_toast_compression_options[] = {
 	{"pglz", TOAST_PGLZ_COMPRESSION, false},
 #ifdef  USE_LZ4
 	{"lz4", TOAST_LZ4_COMPRESSION, false},
+#endif
+#ifdef  USE_ZSTD
+	{"zstd", TOAST_ZSTD_COMPRESSION, false},
 #endif
 	{NULL, 0, false}
 };
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index 34826d01380..4dd6da32324 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -756,7 +756,7 @@ autovacuum_worker_slots = 16	# autovacuum worker slots to allocate
 #row_security = on
 #default_table_access_method = 'heap'
 #default_tablespace = ''		# a tablespace name, '' uses the default
-#default_toast_compression = 'pglz'	# 'pglz' or 'lz4'
+#default_toast_compression = 'pglz'	# 'pglz' or 'lz4' or 'zstd'
 #temp_tablespaces = ''			# a list of tablespace names, '' uses
 					# only default tablespace
 #check_function_bodies = on
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index 1d08268393e..26951f8f890 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -2171,8 +2171,9 @@ describeOneTableDetails(const char *schemaname,
 			/* these strings are literal in our syntax, so not translated. */
 			printTableAddCell(&cont, (compression[0] == 'p' ? "pglz" :
 									  (compression[0] == 'l' ? "lz4" :
-									   (compression[0] == '\0' ? "" :
-										"???"))),
+									   (compression[0] == 'z' ? "zstd" :
+										(compression[0] == '\0' ? "" :
+										 "???")))),
 							  false, false);
 		}
 
diff --git a/src/bin/psql/tab-complete.in.c b/src/bin/psql/tab-complete.in.c
index c916b9299a8..f507f7111c4 100644
--- a/src/bin/psql/tab-complete.in.c
+++ b/src/bin/psql/tab-complete.in.c
@@ -2875,11 +2875,11 @@ match_previous_words(int pattern_id,
 	/* ALTER TABLE ALTER [COLUMN] <foo> SET ( */
 	else if (Matches("ALTER", "TABLE", MatchAny, "ALTER", "COLUMN", MatchAny, "SET", "(") ||
 			 Matches("ALTER", "TABLE", MatchAny, "ALTER", MatchAny, "SET", "("))
-		COMPLETE_WITH("n_distinct", "n_distinct_inherited");
+		COMPLETE_WITH("n_distinct", "n_distinct_inherited", "zstd_level", "zstd_dict_size");
 	/* ALTER TABLE ALTER [COLUMN] <foo> SET COMPRESSION */
 	else if (Matches("ALTER", "TABLE", MatchAny, "ALTER", "COLUMN", MatchAny, "SET", "COMPRESSION") ||
 			 Matches("ALTER", "TABLE", MatchAny, "ALTER", MatchAny, "SET", "COMPRESSION"))
-		COMPLETE_WITH("DEFAULT", "PGLZ", "LZ4");
+		COMPLETE_WITH("DEFAULT", "PGLZ", "LZ4", "ZSTD");
 	/* ALTER TABLE ALTER [COLUMN] <foo> SET EXPRESSION */
 	else if (Matches("ALTER", "TABLE", MatchAny, "ALTER", "COLUMN", MatchAny, "SET", "EXPRESSION") ||
 			 Matches("ALTER", "TABLE", MatchAny, "ALTER", MatchAny, "SET", "EXPRESSION"))
diff --git a/src/include/access/toast_compression.h b/src/include/access/toast_compression.h
index 13c4612ceed..fe4fc3db471 100644
--- a/src/include/access/toast_compression.h
+++ b/src/include/access/toast_compression.h
@@ -13,6 +13,10 @@
 #ifndef TOAST_COMPRESSION_H
 #define TOAST_COMPRESSION_H
 
+#ifdef USE_ZSTD
+#include <zstd.h>
+#endif
+
 /*
  * GUC support.
  *
@@ -38,7 +42,8 @@ typedef enum ToastCompressionId
 {
 	TOAST_PGLZ_COMPRESSION_ID = 0,
 	TOAST_LZ4_COMPRESSION_ID = 1,
-	TOAST_INVALID_COMPRESSION_ID = 2,
+	TOAST_ZSTD_COMPRESSION_ID = 2,
+	TOAST_INVALID_COMPRESSION_ID = 3
 } ToastCompressionId;
 
 /*
@@ -48,10 +53,24 @@ typedef enum ToastCompressionId
  */
 #define TOAST_PGLZ_COMPRESSION			'p'
 #define TOAST_LZ4_COMPRESSION			'l'
+#define TOAST_ZSTD_COMPRESSION			'z'
 #define InvalidCompressionMethod		'\0'
 
 #define CompressionMethodIsValid(cm)  ((cm) != InvalidCompressionMethod)
 
+#define InvalidDictId						0
+
+#ifdef USE_ZSTD
+#define DEFAULT_ZSTD_LEVEL					ZSTD_CLEVEL_DEFAULT
+#define MIN_ZSTD_LEVEL						(int)-ZSTD_BLOCKSIZE_MAX
+#define MAX_ZSTD_LEVEL						22
+#define DEFAULT_ZSTD_DICT_SIZE 				(4 * 1024)	/* 4 KB */
+#else
+#define DEFAULT_ZSTD_LEVEL					0
+#define MIN_ZSTD_LEVEL						0
+#define MAX_ZSTD_LEVEL						0
+#define DEFAULT_ZSTD_DICT_SIZE 				0
+#endif
 
 /* pglz compression/decompression routines */
 extern struct varlena *pglz_compress_datum(const struct varlena *value);
@@ -65,6 +84,11 @@ extern struct varlena *lz4_decompress_datum(const struct varlena *value);
 extern struct varlena *lz4_decompress_datum_slice(const struct varlena *value,
 												  int32 slicelength);
 
+/* zstd compression/decompression routines */
+extern struct varlena *zstd_compress_datum(const struct varlena *value, Oid dictid, int zstd_level);
+extern struct varlena *zstd_decompress_datum(const struct varlena *value);
+extern struct varlena *zstd_decompress_datum_slice(const struct varlena *value, int32 slicelength);
+
 /* other stuff */
 extern ToastCompressionId toast_get_compression_id(struct varlena *attr);
 extern char CompressionNameToMethod(const char *compression);
diff --git a/src/include/access/toast_helper.h b/src/include/access/toast_helper.h
index e6ab8afffb6..08bf3dfc673 100644
--- a/src/include/access/toast_helper.h
+++ b/src/include/access/toast_helper.h
@@ -33,6 +33,8 @@ typedef struct
 	int32		tai_size;
 	uint8		tai_colflags;
 	char		tai_compression;
+	Oid			dictid;
+	int			zstd_level;
 } ToastAttrInfo;
 
 /*
diff --git a/src/include/access/toast_internals.h b/src/include/access/toast_internals.h
index 06ae8583c1e..395afe915d8 100644
--- a/src/include/access/toast_internals.h
+++ b/src/include/access/toast_internals.h
@@ -27,25 +27,54 @@ typedef struct toast_compress_header
 								 * external size; see va_extinfo */
 } toast_compress_header;
 
+typedef struct toast_compress_header_ext
+{
+	int32		vl_len_;		/* varlena header (do not touch directly!) */
+	uint32		tcinfo;			/* 2 bits for compression method and 30 bits
+								 * external size; see va_extinfo */
+	uint32		ext_alg;		/* compression method */
+	Oid			dictid;			/* Dictionary Id */
+}			toast_compress_header_ext;
+
+typedef struct CompressionInfo
+{
+	char		cmethod;
+	Oid			dictid;
+	int			zstd_level;		/* ZSTD compression level */
+}			CompressionInfo;
+
 /*
  * Utilities for manipulation of header information for compressed
  * toast entries.
  */
 #define TOAST_COMPRESS_EXTSIZE(ptr) \
 	(((toast_compress_header *) (ptr))->tcinfo & VARLENA_EXTSIZE_MASK)
-#define TOAST_COMPRESS_METHOD(ptr) \
-	(((toast_compress_header *) (ptr))->tcinfo >> VARLENA_EXTSIZE_BITS)
+#define TOAST_COMPRESS_METHOD(PTR) \
+	( ((((toast_compress_header *) (PTR))->tcinfo >> VARLENA_EXTSIZE_BITS) == VARLENA_EXTENDED_COMPRESSION_FLAG ) \
+		? (((toast_compress_header_ext *) (PTR))->ext_alg) \
+		: ( (((toast_compress_header *) (PTR))->tcinfo) >> VARLENA_EXTSIZE_BITS ) )
 
-#define TOAST_COMPRESS_SET_SIZE_AND_COMPRESS_METHOD(ptr, len, cm_method) \
+#define TOAST_COMPRESS_SET_SIZE_AND_COMPRESS_METHOD(ptr, len, cm_method, dictid) \
 	do { \
 		Assert((len) > 0 && (len) <= VARLENA_EXTSIZE_MASK); \
 		Assert((cm_method) == TOAST_PGLZ_COMPRESSION_ID || \
-			   (cm_method) == TOAST_LZ4_COMPRESSION_ID); \
-		((toast_compress_header *) (ptr))->tcinfo = \
-			(len) | ((uint32) (cm_method) << VARLENA_EXTSIZE_BITS); \
+				(cm_method) == TOAST_LZ4_COMPRESSION_ID || \
+				(cm_method) == TOAST_ZSTD_COMPRESSION_ID); \
+		/* If the compression method is less than TOAST_ZSTD_COMPRESSION_ID, don't use ext_alg */ \
+		if ((cm_method) < TOAST_ZSTD_COMPRESSION_ID) { \
+			((toast_compress_header *) (ptr))->tcinfo = \
+				(len) | ((uint32) (cm_method) << VARLENA_EXTSIZE_BITS); \
+		} else { \
+			/* For compression methods after lz4, use 'VARLENA_EXTENDED_COMPRESSION_FLAG' \
+				in the top bits of tcinfo to indicate compression algorithm is stored in ext_alg. */ \
+			((toast_compress_header_ext *) (ptr))->tcinfo = \
+			(len) | ((uint32)VARLENA_EXTENDED_COMPRESSION_FLAG << VARLENA_EXTSIZE_BITS); \
+			((toast_compress_header_ext *) (ptr))->ext_alg = (cm_method); \
+			((toast_compress_header_ext *) (ptr))->dictid = (dictid); \
+		} \
 	} while (0)
 
-extern Datum toast_compress_datum(Datum value, char cmethod);
+extern Datum toast_compress_datum(Datum value, CompressionInfo cmp);
 extern Oid	toast_get_valid_index(Oid toastoid, LOCKMODE lock);
 
 extern void toast_delete_datum(Relation rel, Datum value, bool is_speculative);
@@ -59,5 +88,6 @@ extern int	toast_open_indexes(Relation toastrel,
 extern void toast_close_indexes(Relation *toastidxs, int num_indexes,
 								LOCKMODE lock);
 extern Snapshot get_toast_snapshot(void);
+extern void setup_compression_info(CompressionInfo * info, Form_pg_attribute att);
 
 #endif							/* TOAST_INTERNALS_H */
diff --git a/src/include/catalog/Makefile b/src/include/catalog/Makefile
index 2bbc7805fe3..1ecd76dd312 100644
--- a/src/include/catalog/Makefile
+++ b/src/include/catalog/Makefile
@@ -81,7 +81,8 @@ CATALOG_HEADERS := \
 	pg_publication_namespace.h \
 	pg_publication_rel.h \
 	pg_subscription.h \
-	pg_subscription_rel.h
+	pg_subscription_rel.h \
+	pg_zstd_dictionaries.h
 
 GENERATED_HEADERS := $(CATALOG_HEADERS:%.h=%_d.h)
 
diff --git a/src/include/catalog/meson.build b/src/include/catalog/meson.build
index ec1cf467f6f..e9cb6d911cc 100644
--- a/src/include/catalog/meson.build
+++ b/src/include/catalog/meson.build
@@ -69,6 +69,7 @@ catalog_headers = [
   'pg_publication_rel.h',
   'pg_subscription.h',
   'pg_subscription_rel.h',
+  'pg_zstd_dictionaries.h',
 ]
 
 # The .dat files we need can just be listed alphabetically.
diff --git a/src/include/catalog/pg_zstd_dictionaries.h b/src/include/catalog/pg_zstd_dictionaries.h
new file mode 100644
index 00000000000..cf847ee2801
--- /dev/null
+++ b/src/include/catalog/pg_zstd_dictionaries.h
@@ -0,0 +1,48 @@
+/*-------------------------------------------------------------------------
+ *
+ * pg_zstd_dictionaries.h
+ *	  definition of the "zstd dictionary" system catalog (pg_zstd_dictionaries)
+ *
+ * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
+ *
+ * src/include/catalog/pg_zstd_dictionaries.h
+ *
+ * NOTES
+ *	  The Catalog.pm module reads this file and derives schema
+ *	  information.
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef PG_ZSTD_DICTIONARIES_H
+#define PG_ZSTD_DICTIONARIES_H
+
+#include "catalog/genbki.h"
+#include "catalog/pg_zstd_dictionaries_d.h"
+
+/* ----------------
+ *		pg_zstd_dictionaries definition.  cpp turns this into
+ *		typedef struct FormData_pg_zstd_dictionaries
+ * ----------------
+ */
+CATALOG(pg_zstd_dictionaries,9946,ZstdDictionariesRelationId)
+{
+	Oid			dictid;
+
+	/*
+	 * variable-length fields start here, but we allow direct access to dict
+	 */
+	bytea		dict BKI_FORCE_NOT_NULL;
+} FormData_pg_zstd_dictionaries;
+
+/* Pointer type to a tuple with the format of pg_zstd_dictionaries relation */
+typedef FormData_pg_zstd_dictionaries *Form_pg_zstd_dictionaries;
+
+DECLARE_TOAST_WITH_MACRO(pg_zstd_dictionaries, 9947, 9948, PgZstdDictionariesToastTable, PgZstdDictionariesToastIndex);
+
+DECLARE_UNIQUE_INDEX_PKEY(pg_zstd_dictionaries_dictid_index, 9949, ZstdDictidIndexId, pg_zstd_dictionaries, btree(dictid oid_ops));
+
+MAKE_SYSCACHE(ZSTDDICTIDOID, pg_zstd_dictionaries_dictid_index, 128);
+
+extern bytea *get_zstd_dict(Oid dictid);
+
+#endif							/* PG_ZSTD_DICTIONARIES_H */
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 4610fc61293..a74c93c020c 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -2366,6 +2366,7 @@ typedef enum ObjectType
 	OBJECT_TYPE,
 	OBJECT_USER_MAPPING,
 	OBJECT_VIEW,
+	OBJECT_ZSTD_DICTIONARY,
 } ObjectType;
 
 /* ----------------------
diff --git a/src/include/utils/attoptcache.h b/src/include/utils/attoptcache.h
index f684a772af5..ee16bf3a4d4 100644
--- a/src/include/utils/attoptcache.h
+++ b/src/include/utils/attoptcache.h
@@ -21,6 +21,12 @@ typedef struct AttributeOpts
 	int32		vl_len_;		/* varlena header (do not touch directly!) */
 	float8		n_distinct;
 	float8		n_distinct_inherited;
+	float8		dictid;			/* Oid is a 32-bit unsigned integer, but
+								 * relopt_int is limited to INT_MAX, so it
+								 * cannot represent the full range of Oid
+								 * values. */
+	int			zstd_dict_size;
+	int			zstd_level;
 } AttributeOpts;
 
 extern AttributeOpts *get_attribute_options(Oid attrelid, int attnum);
diff --git a/src/test/regress/expected/compression.out b/src/test/regress/expected/compression.out
index 4dd9ee7200d..94495388ade 100644
--- a/src/test/regress/expected/compression.out
+++ b/src/test/regress/expected/compression.out
@@ -238,10 +238,11 @@ NOTICE:  merging multiple inherited definitions of column "f1"
 -- test default_toast_compression GUC
 SET default_toast_compression = '';
 ERROR:  invalid value for parameter "default_toast_compression": ""
-HINT:  Available values: pglz, lz4.
+HINT:  Available values: pglz, lz4, zstd.
 SET default_toast_compression = 'I do not exist compression';
 ERROR:  invalid value for parameter "default_toast_compression": "I do not exist compression"
-HINT:  Available values: pglz, lz4.
+HINT:  Available values: pglz, lz4, zstd.
+SET default_toast_compression = 'zstd';
 SET default_toast_compression = 'lz4';
 SET default_toast_compression = 'pglz';
 -- test alter compression method
diff --git a/src/test/regress/expected/compression_1.out b/src/test/regress/expected/compression_1.out
index 7bd7642b4b9..0ce49152176 100644
--- a/src/test/regress/expected/compression_1.out
+++ b/src/test/regress/expected/compression_1.out
@@ -233,6 +233,9 @@ HINT:  Available values: pglz.
 SET default_toast_compression = 'I do not exist compression';
 ERROR:  invalid value for parameter "default_toast_compression": "I do not exist compression"
 HINT:  Available values: pglz.
+SET default_toast_compression = 'zstd';
+ERROR:  invalid value for parameter "default_toast_compression": "zstd"
+HINT:  Available values: pglz.
 SET default_toast_compression = 'lz4';
 ERROR:  invalid value for parameter "default_toast_compression": "lz4"
 HINT:  Available values: pglz.
diff --git a/src/test/regress/sql/compression.sql b/src/test/regress/sql/compression.sql
index 490595fcfb2..e29909558f9 100644
--- a/src/test/regress/sql/compression.sql
+++ b/src/test/regress/sql/compression.sql
@@ -102,6 +102,7 @@ CREATE TABLE cminh() INHERITS (cmdata, cmdata3);
 -- test default_toast_compression GUC
 SET default_toast_compression = '';
 SET default_toast_compression = 'I do not exist compression';
+SET default_toast_compression = 'zstd';
 SET default_toast_compression = 'lz4';
 SET default_toast_compression = 'pglz';
 
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index d16bc208654..86fb79b2076 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -905,6 +905,7 @@ FormData_pg_ts_parser
 FormData_pg_ts_template
 FormData_pg_type
 FormData_pg_user_mapping
+FormData_pg_zstd_dictionaries
 FormExtraData_pg_attribute
 Form_pg_aggregate
 Form_pg_am
@@ -964,6 +965,7 @@ Form_pg_ts_parser
 Form_pg_ts_template
 Form_pg_type
 Form_pg_user_mapping
+Form_pg_zstd_dictionaries
 FormatNode
 FreeBlockNumberArray
 FreeListData
-- 
2.47.1