v1-0001-Add-pg_get_table_ddl-to-reconstruct-CREATE-TABLE.patch

From 8c495d04657841352ce982b28ca9203b2879723e Mon Sep 17 00:00:00 2001
From: Akshay Joshi <akshay.joshi@enterprisedb.com>
Date: Tue, 2 Jun 2026 14:18:40 +0530
Subject: [PATCH v1] Add pg_get_table_ddl() to reconstruct CREATE TABLE
 statements

The function reconstructs the CREATE TABLE statement for an ordinary or
partitioned table, followed by the ALTER TABLE / CREATE INDEX /
CREATE RULE / CREATE STATISTICS statements needed to restore its full
definition.  Each statement is returned as a separate row.

Supported per-column features: data type with type modifiers, COLLATE,
STORAGE, COMPRESSION (pglz / lz4), GENERATED ALWAYS AS (expr)
STORED/VIRTUAL, GENERATED ALWAYS|BY DEFAULT AS IDENTITY (with sequence
options), DEFAULT, NOT NULL, and per-column attoptions emitted as
ALTER COLUMN SET (...).

Supported table-level features: UNLOGGED, INHERITS, PARTITION BY (RANGE
/ LIST / HASH parents), PARTITION OF parent FOR VALUES (FROM/TO, WITH
modulus/remainder, DEFAULT), USING table access method, WITH
(reloptions), TABLESPACE, and inline CHECK constraints in the
CREATE TABLE body.

Supported sub-objects (re-using existing deparse helpers from
ruleutils.c): indexes via pg_get_indexdef_string, constraints
(PRIMARY KEY, UNIQUE, FOREIGN KEY, EXCLUDE, named NOT NULL) via
pg_get_constraintdef_command, rules via pg_get_ruledef, extended
statistics via pg_get_statisticsobjdef_string, REPLICA IDENTITY
NOTHING/FULL/USING INDEX, ALTER TABLE ENABLE/FORCE ROW LEVEL SECURITY,
and child-local DEFAULT overrides on inheritance/partition children.

Default omission convention: every optional clause is dropped when its
value equals what the system would reapply on round-trip, including
type-default COLLATE, per-type STORAGE, the auto-generated identity
sequence name and parameter defaults, heap access method, default
REPLICA IDENTITY, disabled RLS toggles, empty reloptions, and the
default tablespace.

A regression test under src/test/regress covers ordinary tables,
identity (default and custom sequence options), generated columns,
STORAGE/COMPRESSION, constraints (CHECK/UNIQUE/FK with deferrable),
functional and partial indexes, inheritance and partitioning,
partition children with FOR VALUES FROM/TO, WITH modulus/remainder, and
DEFAULT, rules, extended statistics, RLS toggles, REPLICA IDENTITY,
UNLOGGED with reloptions, per-column attoptions, child DEFAULT
overrides, pretty mode, owner=false, and the error paths for views,
sequences, NULL, unknown options, and odd-variadic argument counts.

Author: Akshay Joshi <akshay.joshi@enterprisedb.com>
---
 doc/src/sgml/func/func-info.sgml              |   61 +
 src/backend/commands/tablecmds.c              |   27 +
 src/backend/utils/adt/ddlutils.c              | 1325 +++++++++++++++++
 src/include/catalog/pg_proc.dat               |    7 +
 src/include/commands/tablecmds.h              |    1 +
 .../regress/expected/pg_get_table_ddl.out     |  498 +++++++
 src/test/regress/parallel_schedule            |    2 +-
 src/test/regress/sql/pg_get_table_ddl.sql     |  278 ++++
 8 files changed, 2198 insertions(+), 1 deletion(-)
 create mode 100644 src/test/regress/expected/pg_get_table_ddl.out
 create mode 100644 src/test/regress/sql/pg_get_table_ddl.sql

diff --git a/doc/src/sgml/func/func-info.sgml b/doc/src/sgml/func/func-info.sgml
index 00f64f50ceb..48e987208d5 100644
--- a/doc/src/sgml/func/func-info.sgml
+++ b/doc/src/sgml/func/func-info.sgml
@@ -3961,6 +3961,67 @@ acl      | {postgres=arwdDxtm/postgres,foo=r/postgres}
         <literal>TABLESPACE</literal>.
        </para></entry>
       </row>
+      <row>
+       <entry role="func_table_entry"><para role="func_signature">
+        <indexterm>
+         <primary>pg_get_table_ddl</primary>
+        </indexterm>
+        <function>pg_get_table_ddl</function>
+        ( <parameter>table</parameter> <type>regclass</type>
+        <optional>, <literal>VARIADIC</literal> <parameter>options</parameter>
+        <type>text</type> </optional> )
+        <returnvalue>setof text</returnvalue>
+       </para>
+       <para>
+        Reconstructs the <command>CREATE TABLE</command> statement for the
+        specified ordinary or partitioned table, followed by the
+        <command>ALTER TABLE</command>, <command>CREATE INDEX</command>,
+        <command>CREATE RULE</command>, and <command>CREATE STATISTICS</command>
+        statements needed to recreate the table's columns, constraints,
+        indexes, rules, extended statistics, and row-level security flags.
+        Inherited columns and constraints are emitted by the parent table's
+        DDL and are not duplicated on inheritance children or partitions.
+        Each statement is returned as a separate row.
+        The following options are supported:
+        <literal>pretty</literal> (boolean) for formatted output,
+        <literal>owner</literal> (boolean) to include the
+        <command>ALTER TABLE ... OWNER TO</command> statement,
+        <literal>tablespace</literal> (boolean) to include the
+        <literal>TABLESPACE</literal> clause on the
+        <command>CREATE TABLE</command> statement, and a family of
+        <literal>includes_<replaceable>category</replaceable></literal>
+        booleans that gate emission of optional sub-objects:
+        <literal>includes_indexes</literal>,
+        <literal>includes_constraints</literal>,
+        <literal>includes_rules</literal>,
+        <literal>includes_statistics</literal>,
+        <literal>includes_triggers</literal>,
+        <literal>includes_policies</literal>,
+        <literal>includes_rls</literal> (the
+        <command>ENABLE</command>/<command>FORCE ROW LEVEL SECURITY</command>
+        toggles), and
+        <literal>includes_replica_identity</literal>.  Each of these
+        defaults to <literal>true</literal>.  The
+        <literal>includes_partition</literal> option, which defaults to
+        <literal>false</literal>, controls whether the DDL for partition
+        children is appended after a partitioned-table parent.
+       </para>
+       <para>
+        All three forms of <command>CREATE TABLE</command> are supported:
+        the ordinary column-list form, the typed-table form
+        (<literal>OF <replaceable>type_name</replaceable></literal>, with
+        per-column <literal>WITH OPTIONS</literal> overrides for local
+        defaults, <literal>NOT NULL</literal>, and <literal>CHECK</literal>
+        constraints), and the <literal>PARTITION OF</literal> form.
+        <literal>TEMPORARY</literal> and <literal>UNLOGGED</literal>
+        persistence modes are emitted from
+        <structfield>relpersistence</structfield>.  For temporary tables
+        registered in the current session, the
+        <literal>ON COMMIT DELETE ROWS</literal> and
+        <literal>ON COMMIT DROP</literal> clauses are emitted; the default
+        <literal>ON COMMIT PRESERVE ROWS</literal> is omitted.
+       </para></entry>
+      </row>
      </tbody>
     </tgroup>
    </table>
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index a1845240a98..b02dfb451f8 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -19583,6 +19583,33 @@ remove_on_commit_action(Oid relid)
 	}
 }
 
+/*
+ * Look up the registered ON COMMIT action for a relation.
+ *
+ * Returns ONCOMMIT_NOOP when nothing was registered, which also covers
+ * temporary tables created with the default ON COMMIT PRESERVE ROWS
+ * behavior (register_on_commit_action() skips those, since no action is
+ * needed at commit).  Entries marked for deletion in the current
+ * transaction are ignored.
+ */
+OnCommitAction
+get_on_commit_action(Oid relid)
+{
+	ListCell   *l;
+
+	foreach(l, on_commits)
+	{
+		OnCommitItem *oc = (OnCommitItem *) lfirst(l);
+
+		if (oc->relid != relid)
+			continue;
+		if (oc->deleting_subid != InvalidSubTransactionId)
+			continue;
+		return oc->oncommit;
+	}
+	return ONCOMMIT_NOOP;
+}
+
 /*
  * Perform ON COMMIT actions.
  *
diff --git a/src/backend/utils/adt/ddlutils.c b/src/backend/utils/adt/ddlutils.c
index f32fcd453ef..3761f65e153 100644
--- a/src/backend/utils/adt/ddlutils.c
+++ b/src/backend/utils/adt/ddlutils.c
@@ -21,12 +21,27 @@
 #include "access/genam.h"
 #include "access/htup_details.h"
 #include "access/table.h"
+#include "access/toast_compression.h"
 #include "catalog/pg_auth_members.h"
 #include "catalog/pg_authid.h"
+#include "catalog/pg_class.h"
+#include "catalog/dependency.h"
 #include "catalog/pg_collation.h"
+#include "catalog/pg_constraint.h"
 #include "catalog/pg_database.h"
 #include "catalog/pg_db_role_setting.h"
+#include "catalog/pg_am.h"
+#include "catalog/pg_inherits.h"
+#include "catalog/pg_policy.h"
+#include "catalog/pg_sequence.h"
+#include "catalog/pg_statistic_ext.h"
 #include "catalog/pg_tablespace.h"
+#include "catalog/pg_trigger.h"
+#include "catalog/partition.h"
+#include "commands/defrem.h"
+#include "commands/tablecmds.h"
+#include "rewrite/prs2lock.h"
+#include "nodes/nodes.h"
 #include "commands/tablespace.h"
 #include "common/relpath.h"
 #include "funcapi.h"
@@ -86,6 +101,27 @@ static List *pg_get_tablespace_ddl_internal(Oid tsid, bool pretty, bool no_owner
 static Datum pg_get_tablespace_ddl_srf(FunctionCallInfo fcinfo, Oid tsid, bool isnull);
 static List *pg_get_database_ddl_internal(Oid dbid, bool pretty,
 										  bool no_owner, bool no_tablespace);
+static List *pg_get_table_ddl_internal(Oid relid, bool pretty,
+									   bool no_owner, bool no_tablespace,
+									   bool include_indexes,
+									   bool include_constraints,
+									   bool include_rules,
+									   bool include_statistics,
+									   bool include_triggers,
+									   bool include_policies,
+									   bool include_rls,
+									   bool include_replica_identity,
+									   bool include_partition);
+static void append_column_defs(StringInfo buf, Relation rel, bool pretty,
+							   bool include_constraints);
+static void append_typed_column_overrides(StringInfo buf, Relation rel,
+										  bool pretty, bool include_constraints);
+static void append_inline_check_constraints(StringInfo buf, Relation rel,
+											bool pretty, bool *first);
+static char *find_attrdef_text(Relation rel, AttrNumber attnum,
+							   List **dpcontext);
+static char *lookup_qualified_relname(Oid relid);
+static List *get_inheritance_parents(Oid relid);
 
 
 /*
@@ -1185,3 +1221,1292 @@ pg_get_database_ddl(PG_FUNCTION_ARGS)
 		SRF_RETURN_DONE(funcctx);
 	}
 }
+
+/*
+ * get_inheritance_parents
+ *		Return a List of parent OIDs for relid, ordered by inhseqno.
+ *
+ * find_inheritance_children() walks the opposite direction (parent->children),
+ * so we scan pg_inherits directly here using the (inhrelid, inhseqno) index,
+ * which yields rows in the order they need to appear in the INHERITS clause.
+ * Partition children also have a pg_inherits entry, so callers must skip the
+ * INHERITS clause when relispartition is true.
+ */
+static List *
+get_inheritance_parents(Oid relid)
+{
+	Relation	inheritsRel;
+	SysScanDesc scan;
+	ScanKeyData key;
+	HeapTuple	tup;
+	List	   *parents = NIL;
+
+	inheritsRel = table_open(InheritsRelationId, AccessShareLock);
+	ScanKeyInit(&key,
+				Anum_pg_inherits_inhrelid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(relid));
+	scan = systable_beginscan(inheritsRel, InheritsRelidSeqnoIndexId,
+							  true, NULL, 1, &key);
+	while (HeapTupleIsValid(tup = systable_getnext(scan)))
+	{
+		Form_pg_inherits inh = (Form_pg_inherits) GETSTRUCT(tup);
+
+		parents = lappend_oid(parents, inh->inhparent);
+	}
+	systable_endscan(scan);
+	table_close(inheritsRel, AccessShareLock);
+
+	return parents;
+}
+
+/*
+ * lookup_qualified_relname
+ *		Return the schema-qualified, identifier-quoted name of a relation,
+ *		or raise ERRCODE_UNDEFINED_OBJECT if the relation has disappeared.
+ *
+ * Replaces the unsafe pattern
+ *	  quote_qualified_identifier(get_namespace_name(get_rel_namespace(oid)),
+ *	                             get_rel_name(oid))
+ * which dereferences NULL when a concurrent transaction has dropped the
+ * referenced relation (or its schema) between when we cached its OID and
+ * when we ask the syscache for its name.  Holding AccessShareLock on a
+ * dependent relation makes this race vanishingly unlikely in practice,
+ * but we still defend against it because the alternative is a SIGSEGV.
+ *
+ * Caller is responsible for pfree()ing the result.
+ */
+static char *
+lookup_qualified_relname(Oid relid)
+{
+	HeapTuple	tp;
+	Form_pg_class reltup;
+	char	   *nspname;
+	char	   *result;
+
+	tp = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
+	if (!HeapTupleIsValid(tp))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("relation with OID %u does not exist", relid),
+				 errdetail("It may have been concurrently dropped.")));
+
+	reltup = (Form_pg_class) GETSTRUCT(tp);
+	nspname = get_namespace_name(reltup->relnamespace);
+	if (nspname == NULL)
+	{
+		Oid			nspoid = reltup->relnamespace;
+
+		ReleaseSysCache(tp);
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("schema with OID %u does not exist", nspoid),
+				 errdetail("It may have been concurrently dropped.")));
+	}
+
+	result = quote_qualified_identifier(nspname, NameStr(reltup->relname));
+
+	pfree(nspname);
+	ReleaseSysCache(tp);
+
+	return result;
+}
+
+/*
+ * find_attrdef_text
+ *		Return the deparsed DEFAULT/GENERATED expression for attnum on rel,
+ *		or NULL if no entry exists in TupleConstr->defval.
+ *
+ * The caller passes a List ** so that the deparse context is built lazily
+ * and reused across calls (deparse_context_for is not cheap).  Returned
+ * string is palloc'd in the current memory context; caller pfree's it.
+ */
+static char *
+find_attrdef_text(Relation rel, AttrNumber attnum, List **dpcontext)
+{
+	TupleConstr *constr = RelationGetDescr(rel)->constr;
+
+	if (constr == NULL)
+		return NULL;
+
+	for (int j = 0; j < constr->num_defval; j++)
+	{
+		if (constr->defval[j].adnum != attnum)
+			continue;
+
+		if (*dpcontext == NIL)
+			*dpcontext = deparse_context_for(RelationGetRelationName(rel),
+											 RelationGetRelid(rel));
+
+		return deparse_expression(stringToNode(constr->defval[j].adbin),
+								  *dpcontext, false, false);
+	}
+	return NULL;
+}
+
+/*
+ * append_inline_check_constraints
+ *		Emit each locally-declared CHECK constraint on rel as
+ *		"CONSTRAINT name <pg_get_constraintdef>", separated by ',' from any
+ *		previously-emitted column or constraint.
+ *
+ * *first tracks whether anything has been emitted on this list yet, so the
+ * caller can chain column emission and constraint emission through the same
+ * buffer.  Inherited CHECK constraints (!conislocal) come from the parent's
+ * DDL and aren't repeated here.
+ */
+static void
+append_inline_check_constraints(StringInfo buf, Relation rel, bool pretty,
+								bool *first)
+{
+	Relation	conRel;
+	SysScanDesc conScan;
+	ScanKeyData conKey;
+	HeapTuple	conTup;
+
+	conRel = table_open(ConstraintRelationId, AccessShareLock);
+	ScanKeyInit(&conKey,
+				Anum_pg_constraint_conrelid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(RelationGetRelid(rel)));
+	conScan = systable_beginscan(conRel, ConstraintRelidTypidNameIndexId,
+								 true, NULL, 1, &conKey);
+
+	while (HeapTupleIsValid(conTup = systable_getnext(conScan)))
+	{
+		Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(conTup);
+		Datum		defDatum;
+		char	   *defbody;
+
+		if (con->contype != CONSTRAINT_CHECK)
+			continue;
+		if (!con->conislocal)
+			continue;
+
+		if (!*first)
+			appendStringInfoChar(buf, ',');
+		*first = false;
+		if (pretty)
+			appendStringInfoString(buf, "\n    ");
+		else
+			appendStringInfoChar(buf, ' ');
+
+		defDatum = OidFunctionCall1(F_PG_GET_CONSTRAINTDEF_OID,
+									ObjectIdGetDatum(con->oid));
+		defbody = TextDatumGetCString(defDatum);
+		appendStringInfo(buf, "CONSTRAINT %s %s",
+						 quote_identifier(NameStr(con->conname)),
+						 defbody);
+		pfree(defbody);
+	}
+	systable_endscan(conScan);
+	table_close(conRel, AccessShareLock);
+}
+
+/*
+ * append_column_defs
+ *		Append the comma-separated column definition list for a table.
+ *
+ * Emits each non-dropped, locally-declared column as
+ *		name type [COLLATE x] [STORAGE s] [COMPRESSION c]
+ *		[GENERATED ... | DEFAULT e] [NOT NULL]
+ * followed by any locally-declared inline CHECK constraints.  Optional
+ * clauses are omitted when their value matches what the system would
+ * reapply on round-trip (e.g. type-default COLLATE, type-default STORAGE).
+ */
+static void
+append_column_defs(StringInfo buf, Relation rel, bool pretty,
+				   bool include_constraints)
+{
+	TupleDesc	tupdesc = RelationGetDescr(rel);
+	List	   *dpcontext = NIL;
+	bool		first = true;
+
+	for (int i = 0; i < tupdesc->natts; i++)
+	{
+		Form_pg_attribute att = TupleDescAttr(tupdesc, i);
+		char	   *typstr;
+
+		if (att->attisdropped)
+			continue;
+
+		/*
+		 * Columns inherited from a parent are emitted by the INHERITS clause
+		 * (once implemented), not the column list, unless the child
+		 * redeclared them locally (attislocal=true).
+		 */
+		if (!att->attislocal)
+			continue;
+
+		if (!first)
+			appendStringInfoChar(buf, ',');
+		first = false;
+
+		if (pretty)
+			appendStringInfoString(buf, "\n    ");
+		else
+			appendStringInfoChar(buf, ' ');
+
+		appendStringInfoString(buf, quote_identifier(NameStr(att->attname)));
+		appendStringInfoChar(buf, ' ');
+
+		typstr = format_type_with_typemod(att->atttypid, att->atttypmod);
+		appendStringInfoString(buf, typstr);
+		pfree(typstr);
+
+		/* COLLATE clause, only if it differs from the type's default. */
+		if (OidIsValid(att->attcollation) &&
+			att->attcollation != get_typcollation(att->atttypid))
+			appendStringInfo(buf, " COLLATE %s",
+							 generate_collation_name(att->attcollation));
+
+		/* STORAGE clause, only if it differs from the type's default. */
+		if (att->attstorage != get_typstorage(att->atttypid))
+		{
+			const char *storage = NULL;
+
+			switch (att->attstorage)
+			{
+				case TYPSTORAGE_PLAIN:
+					storage = "PLAIN";
+					break;
+				case TYPSTORAGE_EXTERNAL:
+					storage = "EXTERNAL";
+					break;
+				case TYPSTORAGE_MAIN:
+					storage = "MAIN";
+					break;
+				case TYPSTORAGE_EXTENDED:
+					storage = "EXTENDED";
+					break;
+			}
+			if (storage)
+				appendStringInfo(buf, " STORAGE %s", storage);
+		}
+
+		/* COMPRESSION clause, only if explicitly set on the column. */
+		if (CompressionMethodIsValid(att->attcompression))
+		{
+			const char *cm = NULL;
+
+			switch (att->attcompression)
+			{
+				case TOAST_PGLZ_COMPRESSION:
+					cm = "pglz";
+					break;
+				case TOAST_LZ4_COMPRESSION:
+					cm = "lz4";
+					break;
+			}
+			if (cm)
+				appendStringInfo(buf, " COMPRESSION %s", cm);
+		}
+
+		/*
+		 * Look up the default/generated expression text up front; generated
+		 * columns have atthasdef=true with an entry in pg_attrdef just like
+		 * regular defaults.
+		 */
+		{
+			char	   *defexpr = NULL;
+
+			if (att->atthasdef)
+				defexpr = find_attrdef_text(rel, att->attnum, &dpcontext);
+
+			/* GENERATED / IDENTITY / DEFAULT are mutually exclusive. */
+			if (att->attgenerated == ATTRIBUTE_GENERATED_STORED && defexpr)
+				appendStringInfo(buf, " GENERATED ALWAYS AS (%s) STORED", defexpr);
+			else if (att->attgenerated == ATTRIBUTE_GENERATED_VIRTUAL && defexpr)
+				appendStringInfo(buf, " GENERATED ALWAYS AS (%s) VIRTUAL", defexpr);
+			else if (att->attidentity == ATTRIBUTE_IDENTITY_ALWAYS ||
+					 att->attidentity == ATTRIBUTE_IDENTITY_BY_DEFAULT)
+			{
+				const char *idkind =
+					(att->attidentity == ATTRIBUTE_IDENTITY_ALWAYS)
+					? "ALWAYS" : "BY DEFAULT";
+				Oid			seqid = getIdentitySequence(rel, att->attnum, true);
+
+				appendStringInfo(buf, " GENERATED %s AS IDENTITY", idkind);
+
+				/*
+				 * Emit only the sequence options that differ from their
+				 * defaults — mirroring pg_get_database_ddl's pattern of
+				 * omitting values that the system would reapply on its own.
+				 */
+				if (OidIsValid(seqid))
+				{
+					HeapTuple	seqTup = SearchSysCache1(SEQRELID,
+														 ObjectIdGetDatum(seqid));
+
+					if (HeapTupleIsValid(seqTup))
+					{
+						Form_pg_sequence seq = (Form_pg_sequence) GETSTRUCT(seqTup);
+						StringInfoData opts;
+						bool		first_opt = true;
+						int64		def_min,
+									def_max,
+									def_start;
+						int64		typ_min,
+									typ_max;
+
+						/*
+						 * Per-type bounds for the sequence's underlying
+						 * integer type.  Defaults to int8 if the column type
+						 * is something else (shouldn't happen for IDENTITY,
+						 * but be defensive).
+						 */
+						switch (att->atttypid)
+						{
+							case INT2OID:
+								typ_min = PG_INT16_MIN;
+								typ_max = PG_INT16_MAX;
+								break;
+							case INT4OID:
+								typ_min = PG_INT32_MIN;
+								typ_max = PG_INT32_MAX;
+								break;
+							default:
+								typ_min = PG_INT64_MIN;
+								typ_max = PG_INT64_MAX;
+								break;
+						}
+
+						if (seq->seqincrement > 0)
+						{
+							def_min = 1;
+							def_max = typ_max;
+							def_start = def_min;
+						}
+						else
+						{
+							def_min = typ_min;
+							def_max = -1;
+							def_start = def_max;
+						}
+
+						initStringInfo(&opts);
+
+						/*
+						 * SEQUENCE NAME — omit when it matches the
+						 * implicit "<tablename>_<columnname>_seq" pattern
+						 * in the same schema, since CREATE TABLE will
+						 * regenerate that exact name.  The sequence is an
+						 * INTERNAL dependency of the column, so the lock
+						 * we hold on the table also pins it, but the
+						 * lookup helper still defends against a missing
+						 * pg_class row.
+						 */
+						{
+							HeapTuple	seqClassTup;
+							Form_pg_class seqClass;
+							char		autoname[NAMEDATALEN];
+
+							seqClassTup = SearchSysCache1(RELOID,
+														  ObjectIdGetDatum(seqid));
+							if (!HeapTupleIsValid(seqClassTup))
+								ereport(ERROR,
+										(errcode(ERRCODE_UNDEFINED_OBJECT),
+										 errmsg("identity sequence with OID %u does not exist",
+												seqid),
+										 errdetail("It may have been concurrently dropped.")));
+							seqClass = (Form_pg_class) GETSTRUCT(seqClassTup);
+
+							snprintf(autoname, sizeof(autoname), "%s_%s_seq",
+									 RelationGetRelationName(rel),
+									 NameStr(att->attname));
+							if (seqClass->relnamespace != RelationGetNamespace(rel) ||
+								strcmp(NameStr(seqClass->relname), autoname) != 0)
+							{
+								char	   *seqQual =
+									lookup_qualified_relname(seqid);
+
+								appendStringInfo(&opts, "%sSEQUENCE NAME %s",
+												 first_opt ? "" : " ", seqQual);
+								first_opt = false;
+								pfree(seqQual);
+							}
+							ReleaseSysCache(seqClassTup);
+						}
+
+						if (seq->seqstart != def_start)
+						{
+							appendStringInfo(&opts, "%sSTART WITH " INT64_FORMAT,
+											 first_opt ? "" : " ", seq->seqstart);
+							first_opt = false;
+						}
+						if (seq->seqincrement != 1)
+						{
+							appendStringInfo(&opts, "%sINCREMENT BY " INT64_FORMAT,
+											 first_opt ? "" : " ", seq->seqincrement);
+							first_opt = false;
+						}
+						if (seq->seqmin != def_min)
+						{
+							appendStringInfo(&opts, "%sMINVALUE " INT64_FORMAT,
+											 first_opt ? "" : " ", seq->seqmin);
+							first_opt = false;
+						}
+						if (seq->seqmax != def_max)
+						{
+							appendStringInfo(&opts, "%sMAXVALUE " INT64_FORMAT,
+											 first_opt ? "" : " ", seq->seqmax);
+							first_opt = false;
+						}
+						if (seq->seqcache != 1)
+						{
+							appendStringInfo(&opts, "%sCACHE " INT64_FORMAT,
+											 first_opt ? "" : " ", seq->seqcache);
+							first_opt = false;
+						}
+						if (seq->seqcycle)
+						{
+							appendStringInfo(&opts, "%sCYCLE", first_opt ? "" : " ");
+							first_opt = false;
+						}
+
+						if (!first_opt)
+							appendStringInfo(buf, " (%s)", opts.data);
+
+						pfree(opts.data);
+						ReleaseSysCache(seqTup);
+					}
+				}
+			}
+			else if (defexpr)
+				appendStringInfo(buf, " DEFAULT %s", defexpr);
+
+			if (defexpr)
+				pfree(defexpr);
+		}
+
+		if (att->attnotnull)
+			appendStringInfoString(buf, " NOT NULL");
+	}
+
+	/*
+	 * Table-level CHECK constraints — emitted inline in the CREATE TABLE
+	 * body so they appear alongside the columns (the pg_dump shape).  The
+	 * constraint loop later in pg_get_table_ddl_internal skips CHECK
+	 * constraints to avoid double-emission.
+	 */
+	if (include_constraints)
+		append_inline_check_constraints(buf, rel, pretty, &first);
+}
+
+/*
+ * append_typed_column_overrides
+ *		For a typed table (CREATE TABLE ... OF type_name), append the
+ *		optional "(col WITH OPTIONS ..., ...)" list carrying locally
+ *		applied per-column overrides — DEFAULT, NOT NULL, and any locally
+ *		declared CHECK constraints.
+ *
+ * Columns whose type is fully dictated by reloftype emit nothing.  The
+ * parenthesised list is suppressed entirely when no column needs an
+ * override and there are no locally-declared CHECK constraints, matching
+ * the canonical "CREATE TABLE x OF t;" shape.
+ */
+static void
+append_typed_column_overrides(StringInfo buf, Relation rel, bool pretty,
+							  bool include_constraints)
+{
+	TupleDesc	tupdesc = RelationGetDescr(rel);
+	List	   *dpcontext = NIL;
+	StringInfoData inner;
+	bool		first = true;
+
+	initStringInfo(&inner);
+
+	for (int i = 0; i < tupdesc->natts; i++)
+	{
+		Form_pg_attribute att = TupleDescAttr(tupdesc, i);
+		char	   *defexpr = NULL;
+		bool		has_default;
+		bool		has_notnull;
+
+		if (att->attisdropped)
+			continue;
+
+		if (att->atthasdef)
+			defexpr = find_attrdef_text(rel, att->attnum, &dpcontext);
+
+		has_default = (defexpr != NULL);
+		has_notnull = att->attnotnull;
+
+		if (!has_default && !has_notnull)
+		{
+			if (defexpr)
+				pfree(defexpr);
+			continue;
+		}
+
+		if (!first)
+			appendStringInfoChar(&inner, ',');
+		first = false;
+		if (pretty)
+			appendStringInfoString(&inner, "\n    ");
+		else
+			appendStringInfoChar(&inner, ' ');
+
+		appendStringInfo(&inner, "%s WITH OPTIONS",
+						 quote_identifier(NameStr(att->attname)));
+		if (has_default)
+			appendStringInfo(&inner, " DEFAULT %s", defexpr);
+		if (has_notnull)
+			appendStringInfoString(&inner, " NOT NULL");
+
+		if (defexpr)
+			pfree(defexpr);
+	}
+
+	/*
+	 * Locally-declared CHECK constraints on a typed table belong in the
+	 * column-list parentheses, same as for an untyped table.  The
+	 * out-of-line constraint loop later still skips CHECKs.
+	 */
+	if (include_constraints)
+		append_inline_check_constraints(&inner, rel, pretty, &first);
+
+	if (!first)
+	{
+		appendStringInfoString(buf, " (");
+		appendStringInfoString(buf, inner.data);
+		if (pretty)
+			appendStringInfoString(buf, "\n)");
+		else
+			appendStringInfoChar(buf, ')');
+	}
+	pfree(inner.data);
+}
+
+/*
+ * pg_get_table_ddl_internal
+ *		Generate DDL statements to recreate a regular or partitioned table.
+ *
+ * The first list element is the CREATE TABLE statement.  Subsequent
+ * elements are the ALTER TABLE / CREATE INDEX / CREATE RULE /
+ * CREATE STATISTICS statements needed to restore the table's full
+ * definition.
+ *
+ * Trigger and policy emission are scaffolded but not yet wired up — they
+ * are gated on standalone pg_get_trigger_ddl / pg_get_policy_ddl helpers
+ * landing.
+ */
+static List *
+pg_get_table_ddl_internal(Oid relid, bool pretty,
+						  bool no_owner, bool no_tablespace,
+						  bool include_indexes,
+						  bool include_constraints,
+						  bool include_rules,
+						  bool include_statistics,
+						  bool include_triggers,
+						  bool include_policies,
+						  bool include_rls,
+						  bool include_replica_identity,
+						  bool include_partition)
+{
+	Relation	rel;
+	StringInfoData buf;
+	List	   *statements = NIL;
+	char	   *qualname;
+	char		relkind;
+	char		relpersistence;
+	bool		is_typed;
+	AclResult	aclresult;
+
+	rel = table_open(relid, AccessShareLock);
+
+	relkind = rel->rd_rel->relkind;
+	relpersistence = rel->rd_rel->relpersistence;
+	is_typed = OidIsValid(rel->rd_rel->reloftype);
+
+	/*
+	 * The initial cut only supports ordinary and partitioned tables.  Views,
+	 * matviews, foreign tables, sequences, indexes, composite types, and
+	 * TOAST tables are out of scope for now.
+	 */
+	if (relkind != RELKIND_RELATION && relkind != RELKIND_PARTITIONED_TABLE)
+	{
+		char	   *relname = pstrdup(RelationGetRelationName(rel));
+
+		table_close(rel, AccessShareLock);
+		ereport(ERROR,
+				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+				 errmsg("\"%s\" is not an ordinary or partitioned table",
+						relname)));
+	}
+
+	/* Caller needs SELECT on the table to read its definition. */
+	aclresult = pg_class_aclcheck(relid, GetUserId(), ACL_SELECT);
+	if (aclresult != ACLCHECK_OK)
+		aclcheck_error(aclresult, OBJECT_TABLE,
+					   RelationGetRelationName(rel));
+
+	qualname = lookup_qualified_relname(relid);
+
+	initStringInfo(&buf);
+
+	/* pg_class tuple — for relpartbound and reloptions */
+	{
+		HeapTuple	classtup;
+		Datum		reloptDatum;
+		bool		reloptIsnull;
+
+		classtup = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
+		if (!HeapTupleIsValid(classtup))
+			elog(ERROR, "cache lookup failed for relation %u", relid);
+
+		reloptDatum = SysCacheGetAttr(RELOID, classtup,
+									  Anum_pg_class_reloptions, &reloptIsnull);
+
+		/*
+		 * CREATE [TEMPORARY | UNLOGGED] TABLE qualname ...
+		 *
+		 * Persistence applies uniformly to all three CREATE TABLE forms
+		 * (regular column list, OF type_name, and PARTITION OF).  Only one
+		 * of TEMPORARY / UNLOGGED can be set; the relpersistence catalog
+		 * field is the single source of truth.
+		 */
+		appendStringInfoString(&buf, "CREATE ");
+		if (relpersistence == RELPERSISTENCE_TEMP)
+			appendStringInfoString(&buf, "TEMPORARY ");
+		else if (relpersistence == RELPERSISTENCE_UNLOGGED)
+			appendStringInfoString(&buf, "UNLOGGED ");
+		appendStringInfo(&buf, "TABLE %s", qualname);
+
+		if (rel->rd_rel->relispartition)
+		{
+			/* PARTITION OF parent FOR VALUES ... */
+			Oid			parentOid = get_partition_parent(relid, true);
+			char	   *parentQual = lookup_qualified_relname(parentOid);
+			char	   *parentRelname = get_rel_name(parentOid);
+			Datum		boundDatum;
+			bool		boundIsnull;
+			char	   *forValues = NULL;
+			char	   *boundStr = NULL;
+
+			if (parentRelname == NULL)
+				ereport(ERROR,
+						(errcode(ERRCODE_UNDEFINED_OBJECT),
+						 errmsg("partition parent with OID %u does not exist",
+								parentOid),
+						 errdetail("It may have been concurrently dropped.")));
+
+			boundDatum = SysCacheGetAttr(RELOID, classtup,
+										 Anum_pg_class_relpartbound, &boundIsnull);
+			if (!boundIsnull)
+			{
+				Node	   *boundNode;
+				List	   *dpcontext;
+
+				boundStr = TextDatumGetCString(boundDatum);
+				boundNode = stringToNode(boundStr);
+				dpcontext = deparse_context_for(parentRelname, parentOid);
+				forValues = deparse_expression(boundNode, dpcontext, false, false);
+			}
+
+			appendStringInfo(&buf, " PARTITION OF %s %s",
+							 parentQual, forValues ? forValues : "DEFAULT");
+			if (forValues)
+				pfree(forValues);
+			if (boundStr)
+				pfree(boundStr);
+			pfree(parentQual);
+			pfree(parentRelname);
+
+			/*
+			 * Column-level overrides redeclared on the child are emitted
+			 * out-of-line: NOT NULL/CHECK come through the constraint loop
+			 * (conislocal=true), and DEFAULT comes through the dedicated
+			 * ALTER COLUMN SET DEFAULT pass below.
+			 */
+		}
+		else if (is_typed)
+		{
+			/*
+			 * Typed table: CREATE TABLE name OF type_name [(col WITH
+			 * OPTIONS ...)].  The column list (when present) carries only
+			 * locally-applied overrides — defaults, NOT NULL toggles, and
+			 * locally-declared CHECK constraints — for columns whose type
+			 * is otherwise dictated by reloftype.
+			 */
+			char	   *typname = format_type_be_qualified(rel->rd_rel->reloftype);
+
+			appendStringInfo(&buf, " OF %s", typname);
+			pfree(typname);
+
+			append_typed_column_overrides(&buf, rel, pretty, include_constraints);
+		}
+		else
+		{
+			List	   *parents;
+			ListCell   *lc;
+			bool		first;
+
+			appendStringInfoString(&buf, " (");
+
+			append_column_defs(&buf, rel, pretty, include_constraints);
+
+			if (pretty)
+				appendStringInfoString(&buf, "\n)");
+			else
+				appendStringInfoChar(&buf, ')');
+
+			/* INHERITS (parent1, parent2, ...) — non-partition inheritance only */
+			parents = get_inheritance_parents(relid);
+			if (parents != NIL)
+			{
+				appendStringInfoString(&buf, " INHERITS (");
+				first = true;
+				foreach(lc, parents)
+				{
+					Oid			poid = lfirst_oid(lc);
+					char	   *pname = lookup_qualified_relname(poid);
+
+					if (!first)
+						appendStringInfoString(&buf, ", ");
+					first = false;
+					appendStringInfoString(&buf, pname);
+					pfree(pname);
+				}
+				appendStringInfoChar(&buf, ')');
+				list_free(parents);
+			}
+		}
+
+		/* PARTITION BY — applies whenever this relation is a partitioned table */
+		if (relkind == RELKIND_PARTITIONED_TABLE)
+		{
+			Datum		partkeyDatum;
+			char	   *partkey;
+
+			partkeyDatum = OidFunctionCall1(F_PG_GET_PARTKEYDEF,
+											ObjectIdGetDatum(relid));
+			partkey = TextDatumGetCString(partkeyDatum);
+			appendStringInfo(&buf, " PARTITION BY %s", partkey);
+			pfree(partkey);
+		}
+
+		/*
+		 * USING method — emit only when the table access method differs
+		 * from heap (the cluster default).  Pluggable table AMs have been
+		 * supported since PostgreSQL 12.
+		 */
+		if (OidIsValid(rel->rd_rel->relam) &&
+			rel->rd_rel->relam != HEAP_TABLE_AM_OID)
+		{
+			char	   *amname = get_am_name(rel->rd_rel->relam);
+
+			if (amname != NULL)
+			{
+				appendStringInfo(&buf, " USING %s", quote_identifier(amname));
+				pfree(amname);
+			}
+		}
+
+		/* WITH (reloptions) */
+		if (!reloptIsnull)
+		{
+			appendStringInfoString(&buf, " WITH (");
+			get_reloptions(&buf, reloptDatum);
+			appendStringInfoChar(&buf, ')');
+		}
+
+		ReleaseSysCache(classtup);
+	}
+
+	/* TABLESPACE */
+	if (!no_tablespace && OidIsValid(rel->rd_rel->reltablespace))
+	{
+		char	   *tsname = get_tablespace_name(rel->rd_rel->reltablespace);
+
+		if (tsname != NULL)
+		{
+			appendStringInfo(&buf, " TABLESPACE %s", quote_identifier(tsname));
+			pfree(tsname);
+		}
+	}
+
+	/*
+	 * ON COMMIT applies only to temporary tables.  The action lives in a
+	 * backend-local list (not the catalog) since it's a session-scoped
+	 * property, so this is best-effort: we can only see entries registered
+	 * in the current backend.  PRESERVE ROWS is the default and is not
+	 * emitted; NOOP indicates no entry was found.
+	 */
+	if (relpersistence == RELPERSISTENCE_TEMP)
+	{
+		OnCommitAction oc = get_on_commit_action(relid);
+
+		if (oc == ONCOMMIT_DELETE_ROWS)
+			appendStringInfoString(&buf, " ON COMMIT DELETE ROWS");
+		else if (oc == ONCOMMIT_DROP)
+			appendStringInfoString(&buf, " ON COMMIT DROP");
+	}
+
+	appendStringInfoChar(&buf, ';');
+	statements = lappend(statements, pstrdup(buf.data));
+
+	/* OWNER */
+	if (!no_owner)
+	{
+		char	   *owner = GetUserNameFromId(rel->rd_rel->relowner, false);
+
+		resetStringInfo(&buf);
+		appendStringInfo(&buf, "ALTER TABLE %s OWNER TO %s;",
+						 qualname, quote_identifier(owner));
+		statements = lappend(statements, pstrdup(buf.data));
+		pfree(owner);
+	}
+
+	/*
+	 * Per-column DEFAULT overrides on inherited/partition children.  The
+	 * column list inside CREATE TABLE only emits locally-declared columns
+	 * (attislocal=true), so any default set locally on a column that came
+	 * from a parent table needs to be re-applied with ALTER COLUMN SET
+	 * DEFAULT.  Note: NOT NULL overrides come out through the constraint
+	 * loop (PG 18 stores them as named pg_constraint entries with
+	 * conislocal=true), and CHECK overrides do too.
+	 */
+	{
+		TupleDesc	tupdesc = RelationGetDescr(rel);
+		List	   *dpcontext = NIL;
+
+		for (int i = 0; i < tupdesc->natts; i++)
+		{
+			Form_pg_attribute att = TupleDescAttr(tupdesc, i);
+			char	   *defstr;
+
+			if (att->attisdropped || att->attislocal || !att->atthasdef)
+				continue;
+
+			defstr = find_attrdef_text(rel, att->attnum, &dpcontext);
+			if (defstr == NULL)
+				continue;
+
+			resetStringInfo(&buf);
+			appendStringInfo(&buf,
+							 "ALTER TABLE %s ALTER COLUMN %s SET DEFAULT %s;",
+							 qualname,
+							 quote_identifier(NameStr(att->attname)),
+							 defstr);
+			statements = lappend(statements, pstrdup(buf.data));
+			pfree(defstr);
+		}
+	}
+
+	/*
+	 * Per-column attoptions — these can't be set inline in CREATE TABLE,
+	 * so they come out as ALTER TABLE ... ALTER COLUMN col SET (...) after
+	 * the table is created.  Typical use: n_distinct overrides for the
+	 * planner.
+	 */
+	{
+		TupleDesc	tupdesc = RelationGetDescr(rel);
+
+		for (int i = 0; i < tupdesc->natts; i++)
+		{
+			Form_pg_attribute att = TupleDescAttr(tupdesc, i);
+			HeapTuple	attTup;
+			Datum		optDatum;
+			bool		optIsnull;
+
+			if (att->attisdropped)
+				continue;
+
+			attTup = SearchSysCache2(ATTNUM,
+									 ObjectIdGetDatum(relid),
+									 Int16GetDatum(att->attnum));
+			if (!HeapTupleIsValid(attTup))
+				continue;
+
+			optDatum = SysCacheGetAttr(ATTNUM, attTup,
+									   Anum_pg_attribute_attoptions, &optIsnull);
+			if (!optIsnull)
+			{
+				resetStringInfo(&buf);
+				appendStringInfo(&buf, "ALTER TABLE %s ALTER COLUMN %s SET (",
+								 qualname,
+								 quote_identifier(NameStr(att->attname)));
+				get_reloptions(&buf, optDatum);
+				appendStringInfoString(&buf, ");");
+				statements = lappend(statements, pstrdup(buf.data));
+			}
+			ReleaseSysCache(attTup);
+		}
+	}
+
+	/*
+	 * Indexes — emit a CREATE INDEX for each non-constraint-backed index on
+	 * the table.  Indexes that back PK/UNIQUE/EXCLUDE constraints are
+	 * emitted by the constraint loop below as part of the ALTER TABLE ...
+	 * ADD CONSTRAINT statement, which creates the index implicitly.
+	 */
+	if (include_indexes)
+	{
+		List	   *indexoids = RelationGetIndexList(rel);
+		ListCell   *lc;
+
+		foreach(lc, indexoids)
+		{
+			Oid			idxoid = lfirst_oid(lc);
+			char	   *idxdef;
+
+			if (OidIsValid(get_index_constraint(idxoid)))
+				continue;
+
+			idxdef = pg_get_indexdef_string(idxoid);
+			resetStringInfo(&buf);
+			appendStringInfo(&buf, "%s;", idxdef);
+			statements = lappend(statements, pstrdup(buf.data));
+			pfree(idxdef);
+		}
+		list_free(indexoids);
+	}
+
+	/*
+	 * Constraints — emit an ALTER TABLE ... ADD CONSTRAINT for each
+	 * locally-defined constraint on the table.  Inherited constraints
+	 * (conislocal=false) are produced by the parent's DDL and propagated
+	 * automatically by INHERITS / PARTITION OF, so we skip them here.
+	 */
+	if (include_constraints)
+	{
+		Relation	conRel;
+		SysScanDesc conScan;
+		ScanKeyData conKey;
+		HeapTuple	conTup;
+
+		conRel = table_open(ConstraintRelationId, AccessShareLock);
+		ScanKeyInit(&conKey,
+					Anum_pg_constraint_conrelid,
+					BTEqualStrategyNumber, F_OIDEQ,
+					ObjectIdGetDatum(relid));
+		conScan = systable_beginscan(conRel, ConstraintRelidTypidNameIndexId,
+									 true, NULL, 1, &conKey);
+
+		while (HeapTupleIsValid(conTup = systable_getnext(conScan)))
+		{
+			Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(conTup);
+			char	   *condef;
+
+			if (!con->conislocal)
+				continue;
+			/* CHECK constraints are emitted inline in the column list. */
+			if (con->contype == CONSTRAINT_CHECK)
+				continue;
+
+			condef = pg_get_constraintdef_command(con->oid);
+			resetStringInfo(&buf);
+			appendStringInfo(&buf, "%s;", condef);
+			statements = lappend(statements, pstrdup(buf.data));
+			pfree(condef);
+		}
+		systable_endscan(conScan);
+		table_close(conRel, AccessShareLock);
+	}
+
+	/*
+	 * Rules — emit a CREATE RULE for each cached rewrite rule on the
+	 * relation.  Internal rules (such as the _RETURN rule on views) live on
+	 * views/matviews and don't appear here because we already restricted
+	 * relkind above.
+	 */
+	if (include_rules && rel->rd_rules != NULL)
+	{
+		for (int i = 0; i < rel->rd_rules->numLocks; i++)
+		{
+			Oid			ruleid = rel->rd_rules->rules[i]->ruleId;
+			Datum		ruledef;
+			char	   *ruledef_str;
+
+			ruledef = OidFunctionCall1(F_PG_GET_RULEDEF_OID,
+									   ObjectIdGetDatum(ruleid));
+			ruledef_str = TextDatumGetCString(ruledef);
+			resetStringInfo(&buf);
+			appendStringInfoString(&buf, ruledef_str);
+			statements = lappend(statements, pstrdup(buf.data));
+			pfree(ruledef_str);
+		}
+	}
+
+	/*
+	 * Extended statistics — iterate pg_statistic_ext by stxrelid and emit
+	 * pg_get_statisticsobjdef_string() for each.
+	 */
+	if (include_statistics)
+	{
+		Relation	statRel;
+		SysScanDesc statScan;
+		ScanKeyData statKey;
+		HeapTuple	statTup;
+
+		statRel = table_open(StatisticExtRelationId, AccessShareLock);
+		ScanKeyInit(&statKey,
+					Anum_pg_statistic_ext_stxrelid,
+					BTEqualStrategyNumber, F_OIDEQ,
+					ObjectIdGetDatum(relid));
+		statScan = systable_beginscan(statRel, StatisticExtRelidIndexId,
+									  true, NULL, 1, &statKey);
+
+		while (HeapTupleIsValid(statTup = systable_getnext(statScan)))
+		{
+			Form_pg_statistic_ext stat = (Form_pg_statistic_ext) GETSTRUCT(statTup);
+			char	   *statdef = pg_get_statisticsobjdef_string(stat->oid);
+
+			resetStringInfo(&buf);
+			appendStringInfo(&buf, "%s;", statdef);
+			statements = lappend(statements, pstrdup(buf.data));
+			pfree(statdef);
+		}
+		systable_endscan(statScan);
+		table_close(statRel, AccessShareLock);
+	}
+
+	/*
+	 * REPLICA IDENTITY — emit only when it differs from the default
+	 * ('d' = use primary key).  This affects logical replication
+	 * behavior, so round-trip fidelity matters.
+	 */
+	if (include_replica_identity &&
+		rel->rd_rel->relreplident != REPLICA_IDENTITY_DEFAULT)
+	{
+		resetStringInfo(&buf);
+		switch (rel->rd_rel->relreplident)
+		{
+			case REPLICA_IDENTITY_NOTHING:
+				appendStringInfo(&buf, "ALTER TABLE %s REPLICA IDENTITY NOTHING;",
+								 qualname);
+				statements = lappend(statements, pstrdup(buf.data));
+				break;
+			case REPLICA_IDENTITY_FULL:
+				appendStringInfo(&buf, "ALTER TABLE %s REPLICA IDENTITY FULL;",
+								 qualname);
+				statements = lappend(statements, pstrdup(buf.data));
+				break;
+			case REPLICA_IDENTITY_INDEX:
+				{
+					Oid			replidx = RelationGetReplicaIndex(rel);
+
+					if (OidIsValid(replidx))
+					{
+						char	   *idxname = get_rel_name(replidx);
+
+						if (idxname == NULL)
+							ereport(ERROR,
+									(errcode(ERRCODE_UNDEFINED_OBJECT),
+									 errmsg("replica identity index with OID %u does not exist",
+											replidx),
+									 errdetail("It may have been concurrently dropped.")));
+
+						appendStringInfo(&buf,
+										 "ALTER TABLE %s REPLICA IDENTITY USING INDEX %s;",
+										 qualname,
+										 quote_identifier(idxname));
+						statements = lappend(statements, pstrdup(buf.data));
+						pfree(idxname);
+					}
+				}
+				break;
+		}
+	}
+
+	/* ENABLE / FORCE ROW LEVEL SECURITY */
+	if (include_rls && rel->rd_rel->relrowsecurity)
+	{
+		resetStringInfo(&buf);
+		appendStringInfo(&buf, "ALTER TABLE %s ENABLE ROW LEVEL SECURITY;",
+						 qualname);
+		statements = lappend(statements, pstrdup(buf.data));
+	}
+	if (include_rls && rel->rd_rel->relforcerowsecurity)
+	{
+		resetStringInfo(&buf);
+		appendStringInfo(&buf, "ALTER TABLE %s FORCE ROW LEVEL SECURITY;",
+						 qualname);
+		statements = lappend(statements, pstrdup(buf.data));
+	}
+
+	/*
+	 * Triggers — scaffolding only.  The standalone pg_get_trigger_ddl()
+	 * function (Phil's re-roll) is the intended emission path; once it
+	 * lands the body of this loop becomes a single call into it.  The
+	 * scan structure, lock acquisition, and tgisinternal filter (which
+	 * skips FK-backing and other system-generated triggers) are settled
+	 * here so that change stays minimal.
+	 */
+	if (include_triggers)
+	{
+		Relation	trigRel;
+		SysScanDesc trigScan;
+		ScanKeyData trigKey;
+		HeapTuple	trigTup;
+
+		trigRel = table_open(TriggerRelationId, AccessShareLock);
+		ScanKeyInit(&trigKey,
+					Anum_pg_trigger_tgrelid,
+					BTEqualStrategyNumber, F_OIDEQ,
+					ObjectIdGetDatum(relid));
+		trigScan = systable_beginscan(trigRel, TriggerRelidNameIndexId,
+									  true, NULL, 1, &trigKey);
+		while (HeapTupleIsValid(trigTup = systable_getnext(trigScan)))
+		{
+			Form_pg_trigger trg = (Form_pg_trigger) GETSTRUCT(trigTup);
+
+			if (trg->tgisinternal)
+				continue;
+
+			/* TODO: append pg_get_trigger_ddl(trg->oid) output here. */
+			(void) trg;
+		}
+		systable_endscan(trigScan);
+		table_close(trigRel, AccessShareLock);
+	}
+
+	/*
+	 * Row-level security policies — scaffolding only.  Once
+	 * pg_get_policy_ddl() (already-submitted patch) lands, the body of
+	 * this loop becomes a per-policy call into it.  The ENABLE/FORCE
+	 * ROW LEVEL SECURITY toggles above are the companion catalog flags
+	 * and are already emitted independently of policies.
+	 */
+	if (include_policies)
+	{
+		Relation	polRel;
+		SysScanDesc polScan;
+		ScanKeyData polKey;
+		HeapTuple	polTup;
+
+		polRel = table_open(PolicyRelationId, AccessShareLock);
+		ScanKeyInit(&polKey,
+					Anum_pg_policy_polrelid,
+					BTEqualStrategyNumber, F_OIDEQ,
+					ObjectIdGetDatum(relid));
+		polScan = systable_beginscan(polRel, PolicyPolrelidPolnameIndexId,
+									 true, NULL, 1, &polKey);
+		while (HeapTupleIsValid(polTup = systable_getnext(polScan)))
+		{
+			Form_pg_policy pol = (Form_pg_policy) GETSTRUCT(polTup);
+
+			/* TODO: append pg_get_policy_ddl(relid, polname) output here. */
+			(void) pol;
+		}
+		systable_endscan(polScan);
+		table_close(polRel, AccessShareLock);
+	}
+
+	/*
+	 * Partition children — when include_partition is true and this relation
+	 * is a partitioned-table parent, recursively emit the DDL for each
+	 * direct partition child.  Each child's own DDL handles further levels
+	 * of sub-partitioning through the same recursion.
+	 */
+	if (include_partition && relkind == RELKIND_PARTITIONED_TABLE)
+	{
+		List	   *children = find_inheritance_children(relid, AccessShareLock);
+		ListCell   *lc;
+
+		foreach(lc, children)
+		{
+			Oid			childoid = lfirst_oid(lc);
+			List	   *childstmts;
+
+			childstmts = pg_get_table_ddl_internal(childoid, pretty,
+												   no_owner, no_tablespace,
+												   include_indexes,
+												   include_constraints,
+												   include_rules,
+												   include_statistics,
+												   include_triggers,
+												   include_policies,
+												   include_rls,
+												   include_replica_identity,
+												   include_partition);
+			statements = list_concat(statements, childstmts);
+		}
+		list_free(children);
+	}
+
+	table_close(rel, AccessShareLock);
+	pfree(buf.data);
+	pfree(qualname);
+
+	return statements;
+}
+
+/*
+ * pg_get_table_ddl
+ *		Return DDL to recreate a table as a set of text rows.
+ */
+Datum
+pg_get_table_ddl(PG_FUNCTION_ARGS)
+{
+	FuncCallContext *funcctx;
+	List	   *statements;
+
+	if (SRF_IS_FIRSTCALL())
+	{
+		MemoryContext oldcontext;
+		Oid			relid;
+		DdlOption	opts[] = {
+			{"pretty", DDL_OPT_BOOL},
+			{"owner", DDL_OPT_BOOL},
+			{"tablespace", DDL_OPT_BOOL},
+			{"includes_indexes", DDL_OPT_BOOL},
+			{"includes_constraints", DDL_OPT_BOOL},
+			{"includes_rules", DDL_OPT_BOOL},
+			{"includes_statistics", DDL_OPT_BOOL},
+			{"includes_triggers", DDL_OPT_BOOL},
+			{"includes_policies", DDL_OPT_BOOL},
+			{"includes_rls", DDL_OPT_BOOL},
+			{"includes_replica_identity", DDL_OPT_BOOL},
+			{"includes_partition", DDL_OPT_BOOL},
+		};
+
+		funcctx = SRF_FIRSTCALL_INIT();
+		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+		if (PG_ARGISNULL(0))
+		{
+			MemoryContextSwitchTo(oldcontext);
+			SRF_RETURN_DONE(funcctx);
+		}
+
+		relid = PG_GETARG_OID(0);
+		parse_ddl_options(fcinfo, 1, opts, lengthof(opts));
+
+		statements = pg_get_table_ddl_internal(relid,
+											   opts[0].isset && opts[0].boolval,
+											   opts[1].isset && !opts[1].boolval,
+											   opts[2].isset && !opts[2].boolval,
+											   !opts[3].isset || opts[3].boolval,
+											   !opts[4].isset || opts[4].boolval,
+											   !opts[5].isset || opts[5].boolval,
+											   !opts[6].isset || opts[6].boolval,
+											   !opts[7].isset || opts[7].boolval,
+											   !opts[8].isset || opts[8].boolval,
+											   !opts[9].isset || opts[9].boolval,
+											   !opts[10].isset || opts[10].boolval,
+											   opts[11].isset && opts[11].boolval);
+		funcctx->user_fctx = statements;
+		funcctx->max_calls = list_length(statements);
+
+		MemoryContextSwitchTo(oldcontext);
+	}
+
+	funcctx = SRF_PERCALL_SETUP();
+	statements = (List *) funcctx->user_fctx;
+
+	if (funcctx->call_cntr < funcctx->max_calls)
+	{
+		char	   *stmt;
+
+		stmt = list_nth(statements, funcctx->call_cntr);
+
+		SRF_RETURN_NEXT(funcctx, CStringGetTextDatum(stmt));
+	}
+	else
+	{
+		list_free_deep(statements);
+		SRF_RETURN_DONE(funcctx);
+	}
+}
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index be157a5fbe9..0d57081a1f3 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -8615,6 +8615,13 @@
   proargtypes => 'regdatabase text', proallargtypes => '{regdatabase,text}',
   proargmodes => '{i,v}', proargdefaults => '{NULL}',
   prosrc => 'pg_get_database_ddl' },
+{ oid => '8215', descr => 'get DDL to recreate a table',
+  proname => 'pg_get_table_ddl', prorows => '50', provariadic => 'text',
+  proisstrict => 'f', proretset => 't', provolatile => 's',
+  pronargdefaults => '1', prorettype => 'text',
+  proargtypes => 'regclass text', proallargtypes => '{regclass,text}',
+  proargmodes => '{i,v}', proargdefaults => '{NULL}',
+  prosrc => 'pg_get_table_ddl' },
 { oid => '2509',
   descr => 'deparse an encoded expression with pretty-print option',
   proname => 'pg_get_expr', provolatile => 's', prorettype => 'text',
diff --git a/src/include/commands/tablecmds.h b/src/include/commands/tablecmds.h
index c3d8518cb62..2085027c9b1 100644
--- a/src/include/commands/tablecmds.h
+++ b/src/include/commands/tablecmds.h
@@ -92,6 +92,7 @@ extern void check_of_type(HeapTuple typetuple);
 
 extern void register_on_commit_action(Oid relid, OnCommitAction action);
 extern void remove_on_commit_action(Oid relid);
+extern OnCommitAction get_on_commit_action(Oid relid);
 
 extern void PreCommit_on_commit_actions(void);
 extern void AtEOXact_on_commit_actions(bool isCommit);
diff --git a/src/test/regress/expected/pg_get_table_ddl.out b/src/test/regress/expected/pg_get_table_ddl.out
new file mode 100644
index 00000000000..a5fc8add515
--- /dev/null
+++ b/src/test/regress/expected/pg_get_table_ddl.out
@@ -0,0 +1,498 @@
+--
+-- pg_get_table_ddl
+--
+-- All tests pass owner=>false so the ALTER TABLE OWNER TO line is not
+-- emitted, keeping output stable across test runners (which may run under
+-- different role names).
+--
+CREATE SCHEMA pgtbl_ddl_test;
+SET search_path = pgtbl_ddl_test;
+-- Basic table with PRIMARY KEY, NOT NULL, DEFAULT, COLLATE.
+CREATE TABLE basic (
+    id int PRIMARY KEY,
+    name text NOT NULL DEFAULT 'anon' COLLATE "C"
+);
+SELECT * FROM pg_get_table_ddl('basic'::regclass, 'owner', 'false');
+                                                pg_get_table_ddl                                                
+----------------------------------------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.basic ( id integer NOT NULL, name text COLLATE "C" DEFAULT 'anon'::text NOT NULL);
+ ALTER TABLE pgtbl_ddl_test.basic ADD CONSTRAINT basic_id_not_null NOT NULL id;
+ ALTER TABLE pgtbl_ddl_test.basic ADD CONSTRAINT basic_name_not_null NOT NULL name;
+ ALTER TABLE pgtbl_ddl_test.basic ADD CONSTRAINT basic_pkey PRIMARY KEY (id);
+(4 rows)
+
+-- Identity columns (with default and custom sequence options).
+CREATE TABLE id_cols (
+    id_always int GENERATED ALWAYS AS IDENTITY,
+    id_default int GENERATED BY DEFAULT AS IDENTITY
+);
+SELECT * FROM pg_get_table_ddl('id_cols'::regclass, 'owner', 'false');
+                                                                       pg_get_table_ddl                                                                        
+---------------------------------------------------------------------------------------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.id_cols ( id_always integer GENERATED ALWAYS AS IDENTITY NOT NULL, id_default integer GENERATED BY DEFAULT AS IDENTITY NOT NULL);
+ ALTER TABLE pgtbl_ddl_test.id_cols ADD CONSTRAINT id_cols_id_always_not_null NOT NULL id_always;
+ ALTER TABLE pgtbl_ddl_test.id_cols ADD CONSTRAINT id_cols_id_default_not_null NOT NULL id_default;
+(3 rows)
+
+CREATE TABLE id_custom (
+    v int GENERATED ALWAYS AS IDENTITY (
+        SEQUENCE NAME id_custom_v_seq
+        START WITH 100 INCREMENT BY 5
+        MINVALUE 50 MAXVALUE 1000
+        CACHE 10 CYCLE
+    )
+);
+SELECT * FROM pg_get_table_ddl('id_custom'::regclass, 'owner', 'false');
+                                                                          pg_get_table_ddl                                                                          
+--------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.id_custom ( v integer GENERATED ALWAYS AS IDENTITY (START WITH 100 INCREMENT BY 5 MINVALUE 50 MAXVALUE 1000 CACHE 10 CYCLE) NOT NULL);
+ ALTER TABLE pgtbl_ddl_test.id_custom ADD CONSTRAINT id_custom_v_not_null NOT NULL v;
+(2 rows)
+
+-- Generated stored column.
+CREATE TABLE gen_cols (
+    cents int,
+    dollars numeric GENERATED ALWAYS AS (cents / 100.0) STORED
+);
+SELECT * FROM pg_get_table_ddl('gen_cols'::regclass, 'owner', 'false');
+                                                        pg_get_table_ddl                                                         
+---------------------------------------------------------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.gen_cols ( cents integer, dollars numeric GENERATED ALWAYS AS (((cents)::numeric / 100.0)) STORED);
+(1 row)
+
+-- STORAGE and COMPRESSION (only emitted when non-default for the type).
+CREATE TABLE storage_cols (
+    a text STORAGE EXTERNAL,
+    b text STORAGE MAIN,
+    c text COMPRESSION pglz
+);
+SELECT * FROM pg_get_table_ddl('storage_cols'::regclass, 'owner', 'false');
+                                                  pg_get_table_ddl                                                  
+--------------------------------------------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.storage_cols ( a text STORAGE EXTERNAL, b text STORAGE MAIN, c text COMPRESSION pglz);
+(1 row)
+
+-- Constraints: CHECK, UNIQUE, FOREIGN KEY (DEFERRABLE).
+CREATE TABLE refd (id int PRIMARY KEY);
+CREATE TABLE cons (
+    a int CHECK (a > 0),
+    b int UNIQUE,
+    c int REFERENCES refd(id) DEFERRABLE INITIALLY DEFERRED
+);
+SELECT * FROM pg_get_table_ddl('cons'::regclass, 'owner', 'false');
+                                                       pg_get_table_ddl                                                        
+-------------------------------------------------------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.cons ( a integer, b integer, c integer, CONSTRAINT cons_a_check CHECK ((a > 0)));
+ ALTER TABLE pgtbl_ddl_test.cons ADD CONSTRAINT cons_b_key UNIQUE (b);
+ ALTER TABLE pgtbl_ddl_test.cons ADD CONSTRAINT cons_c_fkey FOREIGN KEY (c) REFERENCES refd(id) DEFERRABLE INITIALLY DEFERRED;
+(3 rows)
+
+-- Indexes: functional and partial.  Constraint-backing indexes are
+-- suppressed (they are emitted by the constraint loop).
+CREATE TABLE idxd (id int PRIMARY KEY, name text);
+CREATE INDEX idxd_lower ON idxd (lower(name));
+CREATE INDEX idxd_partial ON idxd (id) WHERE id > 100;
+SELECT * FROM pg_get_table_ddl('idxd'::regclass, 'owner', 'false');
+                                  pg_get_table_ddl                                   
+-------------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.idxd ( id integer NOT NULL, name text);
+ CREATE INDEX idxd_lower ON pgtbl_ddl_test.idxd USING btree (lower(name));
+ CREATE INDEX idxd_partial ON pgtbl_ddl_test.idxd USING btree (id) WHERE (id > 100);
+ ALTER TABLE pgtbl_ddl_test.idxd ADD CONSTRAINT idxd_id_not_null NOT NULL id;
+ ALTER TABLE pgtbl_ddl_test.idxd ADD CONSTRAINT idxd_pkey PRIMARY KEY (id);
+(5 rows)
+
+-- Inheritance, including a child DEFAULT override on an inherited column.
+CREATE TABLE par (a int DEFAULT 1, b text);
+CREATE TABLE ch (c int) INHERITS (par);
+ALTER TABLE ch ALTER COLUMN a SET DEFAULT 999;
+SELECT * FROM pg_get_table_ddl('ch'::regclass, 'owner', 'false');
+                              pg_get_table_ddl                              
+----------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.ch ( c integer) INHERITS (pgtbl_ddl_test.par);
+ ALTER TABLE pgtbl_ddl_test.ch ALTER COLUMN a SET DEFAULT 999;
+(2 rows)
+
+-- Per-column attoptions: emitted as ALTER COLUMN SET (...).
+CREATE TABLE attopt (a int, b text);
+ALTER TABLE attopt ALTER COLUMN a SET (n_distinct = 100);
+SELECT * FROM pg_get_table_ddl('attopt'::regclass, 'owner', 'false');
+                             pg_get_table_ddl                             
+--------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.attopt ( a integer, b text);
+ ALTER TABLE pgtbl_ddl_test.attopt ALTER COLUMN a SET (n_distinct='100');
+(2 rows)
+
+-- Partitioned table parent (RANGE and HASH).
+CREATE TABLE parted_range (id int, k int) PARTITION BY RANGE (id);
+SELECT * FROM pg_get_table_ddl('parted_range'::regclass, 'owner', 'false');
+                                      pg_get_table_ddl                                      
+--------------------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.parted_range ( id integer, k integer) PARTITION BY RANGE (id);
+(1 row)
+
+CREATE TABLE parted_hash (id int) PARTITION BY HASH (id);
+SELECT * FROM pg_get_table_ddl('parted_hash'::regclass, 'owner', 'false');
+                               pg_get_table_ddl                                
+-------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.parted_hash ( id integer) PARTITION BY HASH (id);
+(1 row)
+
+-- Partition children: FROM/TO, WITH (modulus, remainder), DEFAULT.
+CREATE TABLE parted_range_1 PARTITION OF parted_range
+    FOR VALUES FROM (0) TO (100);
+ALTER TABLE parted_range_1 ALTER COLUMN k SET DEFAULT 7;
+SELECT * FROM pg_get_table_ddl('parted_range_1'::regclass, 'owner', 'false');
+                                                 pg_get_table_ddl                                                  
+-------------------------------------------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.parted_range_1 PARTITION OF pgtbl_ddl_test.parted_range FOR VALUES FROM (0) TO (100);
+ ALTER TABLE pgtbl_ddl_test.parted_range_1 ALTER COLUMN k SET DEFAULT 7;
+(2 rows)
+
+CREATE TABLE parted_hash_0 PARTITION OF parted_hash
+    FOR VALUES WITH (modulus 2, remainder 0);
+SELECT * FROM pg_get_table_ddl('parted_hash_0'::regclass, 'owner', 'false');
+                                                      pg_get_table_ddl                                                       
+-----------------------------------------------------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.parted_hash_0 PARTITION OF pgtbl_ddl_test.parted_hash FOR VALUES WITH (modulus 2, remainder 0);
+(1 row)
+
+CREATE TABLE parted_range_def PARTITION OF parted_range DEFAULT;
+SELECT * FROM pg_get_table_ddl('parted_range_def'::regclass, 'owner', 'false');
+                                        pg_get_table_ddl                                        
+------------------------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.parted_range_def PARTITION OF pgtbl_ddl_test.parted_range DEFAULT;
+(1 row)
+
+-- Rules.
+CREATE TABLE rt (id int);
+CREATE TABLE rt_log (id int);
+CREATE RULE rt_log_insert AS ON INSERT TO rt
+    DO ALSO INSERT INTO rt_log VALUES (NEW.id);
+SELECT * FROM pg_get_table_ddl('rt'::regclass, 'owner', 'false');
+                        pg_get_table_ddl                        
+----------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.rt ( id integer);
+ CREATE RULE rt_log_insert AS                                  +
+     ON INSERT TO pgtbl_ddl_test.rt DO  INSERT INTO rt_log (id)+
+   VALUES (new.id);
+(2 rows)
+
+-- Extended statistics.
+CREATE TABLE stx (a int, b int, c int);
+CREATE STATISTICS stx_ndv (ndistinct) ON a, b FROM stx;
+SELECT * FROM pg_get_table_ddl('stx'::regclass, 'owner', 'false');
+                            pg_get_table_ddl                            
+------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.stx ( a integer, b integer, c integer);
+ CREATE STATISTICS pgtbl_ddl_test.stx_ndv (ndistinct) ON a, b FROM stx;
+(2 rows)
+
+-- Row-level security toggles.
+CREATE TABLE rls (id int);
+ALTER TABLE rls ENABLE ROW LEVEL SECURITY;
+ALTER TABLE rls FORCE ROW LEVEL SECURITY;
+SELECT * FROM pg_get_table_ddl('rls'::regclass, 'owner', 'false');
+                     pg_get_table_ddl                      
+-----------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.rls ( id integer);
+ ALTER TABLE pgtbl_ddl_test.rls ENABLE ROW LEVEL SECURITY;
+ ALTER TABLE pgtbl_ddl_test.rls FORCE ROW LEVEL SECURITY;
+(3 rows)
+
+-- REPLICA IDENTITY: emitted only when not the default.
+CREATE TABLE ri_full (a int);
+ALTER TABLE ri_full REPLICA IDENTITY FULL;
+SELECT * FROM pg_get_table_ddl('ri_full'::regclass, 'owner', 'false');
+                     pg_get_table_ddl                      
+-----------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.ri_full ( a integer);
+ ALTER TABLE pgtbl_ddl_test.ri_full REPLICA IDENTITY FULL;
+(2 rows)
+
+CREATE TABLE ri_idx (a int NOT NULL);
+CREATE UNIQUE INDEX ri_idx_a ON ri_idx (a);
+ALTER TABLE ri_idx REPLICA IDENTITY USING INDEX ri_idx_a;
+SELECT * FROM pg_get_table_ddl('ri_idx'::regclass, 'owner', 'false');
+                                pg_get_table_ddl                                
+--------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.ri_idx ( a integer NOT NULL);
+ CREATE UNIQUE INDEX ri_idx_a ON pgtbl_ddl_test.ri_idx USING btree (a);
+ ALTER TABLE pgtbl_ddl_test.ri_idx ADD CONSTRAINT ri_idx_a_not_null NOT NULL a;
+ ALTER TABLE pgtbl_ddl_test.ri_idx REPLICA IDENTITY USING INDEX ri_idx_a;
+(4 rows)
+
+-- UNLOGGED + reloptions.
+CREATE UNLOGGED TABLE uno (id int) WITH (fillfactor = 70);
+SELECT * FROM pg_get_table_ddl('uno'::regclass, 'owner', 'false');
+                                pg_get_table_ddl                                
+--------------------------------------------------------------------------------
+ CREATE UNLOGGED TABLE pgtbl_ddl_test.uno ( id integer) WITH (fillfactor='70');
+(1 row)
+
+-- Typed table (CREATE TABLE OF type_name).  Columns inherited from the
+-- type emit nothing; locally-applied DEFAULT, NOT NULL and CHECK come
+-- out through a single "(col WITH OPTIONS ...)" list.
+CREATE TYPE typed_t AS (a int, b text);
+CREATE TABLE typed_plain OF typed_t;
+SELECT * FROM pg_get_table_ddl('typed_plain'::regclass, 'owner', 'false');
+                          pg_get_table_ddl                          
+--------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.typed_plain OF pgtbl_ddl_test.typed_t;
+(1 row)
+
+CREATE TABLE typed_over OF typed_t (
+    a WITH OPTIONS DEFAULT 7 NOT NULL,
+    b WITH OPTIONS NOT NULL,
+    CONSTRAINT b_nonempty CHECK (length(b) > 0)
+);
+SELECT * FROM pg_get_table_ddl('typed_over'::regclass, 'owner', 'false');
+                                                                                pg_get_table_ddl                                                                                
+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.typed_over OF pgtbl_ddl_test.typed_t ( a WITH OPTIONS DEFAULT 7 NOT NULL, b WITH OPTIONS NOT NULL, CONSTRAINT b_nonempty CHECK ((length(b) > 0)));
+ ALTER TABLE pgtbl_ddl_test.typed_over ADD CONSTRAINT typed_over_a_not_null NOT NULL a;
+ ALTER TABLE pgtbl_ddl_test.typed_over ADD CONSTRAINT typed_over_b_not_null NOT NULL b;
+(3 rows)
+
+-- Temporary tables + ON COMMIT.  Temp tables live in a session-local
+-- namespace whose name varies between runs, so we strip the schema
+-- prefix.  ON COMMIT DROP only fires at commit, so the queries that
+-- need to see the catalog entry must run in the same transaction as
+-- the CREATE.
+BEGIN;
+CREATE TEMP TABLE temp_default (id int);
+CREATE TEMP TABLE temp_delete (id int) ON COMMIT DELETE ROWS;
+CREATE TEMP TABLE temp_drop (id int) ON COMMIT DROP;
+SELECT regexp_replace(line, '"?pg_temp_?[0-9]+"?\.', 'pg_temp.') AS line
+FROM pg_get_table_ddl('temp_default'::regclass, 'owner', 'false') AS line
+WHERE line LIKE 'CREATE %';
+                            line                            
+------------------------------------------------------------
+ CREATE TEMPORARY TABLE pg_temp.temp_default ( id integer);
+(1 row)
+
+SELECT regexp_replace(line, '"?pg_temp_?[0-9]+"?\.', 'pg_temp.') AS line
+FROM pg_get_table_ddl('temp_delete'::regclass, 'owner', 'false') AS line
+WHERE line LIKE 'CREATE %';
+                                      line                                       
+---------------------------------------------------------------------------------
+ CREATE TEMPORARY TABLE pg_temp.temp_delete ( id integer) ON COMMIT DELETE ROWS;
+(1 row)
+
+SELECT regexp_replace(line, '"?pg_temp_?[0-9]+"?\.', 'pg_temp.') AS line
+FROM pg_get_table_ddl('temp_drop'::regclass, 'owner', 'false') AS line
+WHERE line LIKE 'CREATE %';
+                                  line                                  
+------------------------------------------------------------------------
+ CREATE TEMPORARY TABLE pg_temp.temp_drop ( id integer) ON COMMIT DROP;
+(1 row)
+
+ROLLBACK;
+-- Pretty mode.
+SELECT * FROM pg_get_table_ddl('basic'::regclass, 'owner', 'false', 'pretty', 'true');
+                                  pg_get_table_ddl                                  
+------------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.basic (                                               +
+     id integer NOT NULL,                                                          +
+     name text COLLATE "C" DEFAULT 'anon'::text NOT NULL                           +
+ );
+ ALTER TABLE pgtbl_ddl_test.basic ADD CONSTRAINT basic_id_not_null NOT NULL id;
+ ALTER TABLE pgtbl_ddl_test.basic ADD CONSTRAINT basic_name_not_null NOT NULL name;
+ ALTER TABLE pgtbl_ddl_test.basic ADD CONSTRAINT basic_pkey PRIMARY KEY (id);
+(4 rows)
+
+-- includes_* gating: each sub-object category can be suppressed individually.
+-- includes_indexes=false hides the CREATE INDEX statements.
+SELECT * FROM pg_get_table_ddl('idxd'::regclass, 'owner', 'false',
+                               'includes_indexes', 'false');
+                               pg_get_table_ddl                               
+------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.idxd ( id integer NOT NULL, name text);
+ ALTER TABLE pgtbl_ddl_test.idxd ADD CONSTRAINT idxd_id_not_null NOT NULL id;
+ ALTER TABLE pgtbl_ddl_test.idxd ADD CONSTRAINT idxd_pkey PRIMARY KEY (id);
+(3 rows)
+
+-- includes_constraints=false hides both inline CHECK in CREATE TABLE and
+-- the ALTER TABLE ... ADD CONSTRAINT lines.
+SELECT * FROM pg_get_table_ddl('cons'::regclass, 'owner', 'false',
+                               'includes_constraints', 'false');
+                           pg_get_table_ddl                           
+----------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.cons ( a integer, b integer, c integer);
+(1 row)
+
+-- includes_rules=false hides the CREATE RULE.
+SELECT * FROM pg_get_table_ddl('rt'::regclass, 'owner', 'false',
+                               'includes_rules', 'false');
+               pg_get_table_ddl                
+-----------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.rt ( id integer);
+(1 row)
+
+-- includes_statistics=false hides the CREATE STATISTICS.
+SELECT * FROM pg_get_table_ddl('stx'::regclass, 'owner', 'false',
+                               'includes_statistics', 'false');
+                          pg_get_table_ddl                           
+---------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.stx ( a integer, b integer, c integer);
+(1 row)
+
+-- includes_rls=false hides the ENABLE/FORCE ROW LEVEL SECURITY toggles.
+SELECT * FROM pg_get_table_ddl('rls'::regclass, 'owner', 'false',
+                               'includes_rls', 'false');
+                pg_get_table_ddl                
+------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.rls ( id integer);
+(1 row)
+
+-- includes_replica_identity=false hides the REPLICA IDENTITY clause.
+SELECT * FROM pg_get_table_ddl('ri_full'::regclass, 'owner', 'false',
+                               'includes_replica_identity', 'false');
+                 pg_get_table_ddl                  
+---------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.ri_full ( a integer);
+(1 row)
+
+-- includes_partition: default is false, so the partitioned-table parent
+-- DDL on its own does not include its children.  Setting it to true
+-- appends the children's DDL after the parent.
+SELECT * FROM pg_get_table_ddl('parted_range'::regclass, 'owner', 'false',
+                               'includes_partition', 'true');
+                                                 pg_get_table_ddl                                                  
+-------------------------------------------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.parted_range ( id integer, k integer) PARTITION BY RANGE (id);
+ CREATE TABLE pgtbl_ddl_test.parted_range_1 PARTITION OF pgtbl_ddl_test.parted_range FOR VALUES FROM (0) TO (100);
+ ALTER TABLE pgtbl_ddl_test.parted_range_1 ALTER COLUMN k SET DEFAULT 7;
+ CREATE TABLE pgtbl_ddl_test.parted_range_def PARTITION OF pgtbl_ddl_test.parted_range DEFAULT;
+(4 rows)
+
+-- Error: not an ordinary or partitioned table.
+CREATE VIEW v AS SELECT 1 AS x;
+SELECT * FROM pg_get_table_ddl('v'::regclass);
+ERROR:  "v" is not an ordinary or partitioned table
+CREATE SEQUENCE s;
+SELECT * FROM pg_get_table_ddl('s'::regclass);
+ERROR:  "s" is not an ordinary or partitioned table
+-- NULL argument returns no rows.
+SELECT * FROM pg_get_table_ddl(NULL);
+ pg_get_table_ddl 
+------------------
+(0 rows)
+
+-- Unrecognized option.
+SELECT * FROM pg_get_table_ddl('basic'::regclass, 'bogus', 'true');
+ERROR:  unrecognized option: "bogus"
+-- Odd number of variadic args.
+SELECT * FROM pg_get_table_ddl('basic'::regclass, 'pretty');
+ERROR:  variadic arguments must be name/value pairs
+HINT:  Provide an even number of variadic arguments that can be divided into pairs.
+-- Round-trip verification.  For every test table, capture the DDL the
+-- function emits, drop the schema, replay each table's DDL in
+-- dependency order, and confirm that pg_get_table_ddl on the recreated
+-- relation matches the original line-for-line.  The final SELECT must
+-- return zero rows.
+CREATE TEMP TABLE pgtbl_ddl_rt_orig (name text, ord int, line text);
+INSERT INTO pgtbl_ddl_rt_orig
+SELECT t.name, o.ord, o.line
+FROM (VALUES
+        ('basic'), ('id_cols'), ('id_custom'), ('gen_cols'), ('storage_cols'),
+        ('refd'), ('cons'), ('idxd'),
+        ('par'), ('ch'), ('attopt'),
+        ('parted_range'), ('parted_range_1'), ('parted_range_def'),
+        ('parted_hash'), ('parted_hash_0'),
+        ('rt_log'), ('rt'),
+        ('stx'), ('rls'), ('ri_full'), ('ri_idx'), ('uno')
+     ) AS t(name),
+     LATERAL pg_get_table_ddl(('pgtbl_ddl_test.' || t.name)::regclass,
+                              'owner', 'false') WITH ORDINALITY o(line, ord);
+DO $$
+DECLARE
+    tables CONSTANT text[] := ARRAY[
+        'basic', 'id_cols', 'id_custom', 'gen_cols', 'storage_cols',
+        'refd', 'cons', 'idxd',
+        'par', 'ch', 'attopt',
+        'parted_range', 'parted_range_1', 'parted_range_def',
+        'parted_hash', 'parted_hash_0',
+        'rt_log', 'rt',
+        'stx', 'rls', 'ri_full', 'ri_idx', 'uno'
+    ];
+    t text;
+    stmt text;
+BEGIN
+    DROP SCHEMA pgtbl_ddl_test CASCADE;
+    CREATE SCHEMA pgtbl_ddl_test;
+    FOREACH t IN ARRAY tables LOOP
+        FOR stmt IN
+            SELECT line FROM pgtbl_ddl_rt_orig WHERE name = t ORDER BY ord
+        LOOP
+            EXECUTE stmt;
+        END LOOP;
+    END LOOP;
+END $$;
+NOTICE:  drop cascades to 25 other objects
+DETAIL:  drop cascades to table basic
+drop cascades to table id_cols
+drop cascades to table id_custom
+drop cascades to table gen_cols
+drop cascades to table storage_cols
+drop cascades to table refd
+drop cascades to table cons
+drop cascades to table idxd
+drop cascades to table par
+drop cascades to table ch
+drop cascades to table attopt
+drop cascades to table parted_range
+drop cascades to table parted_hash
+drop cascades to table rt
+drop cascades to table rt_log
+drop cascades to table stx
+drop cascades to table rls
+drop cascades to table ri_full
+drop cascades to table ri_idx
+drop cascades to table uno
+drop cascades to type typed_t
+drop cascades to table typed_plain
+drop cascades to table typed_over
+drop cascades to view v
+drop cascades to sequence s
+WITH after_ddl AS (
+    SELECT t.name, o.ord, o.line
+    FROM (SELECT DISTINCT name FROM pgtbl_ddl_rt_orig) AS t,
+         LATERAL pg_get_table_ddl(('pgtbl_ddl_test.' || t.name)::regclass,
+                                  'owner', 'false') WITH ORDINALITY o(line, ord)
+)
+(SELECT 'missing-in-copy' AS kind, name, ord, line FROM pgtbl_ddl_rt_orig
+ EXCEPT
+ SELECT 'missing-in-copy', name, ord, line FROM after_ddl)
+UNION ALL
+(SELECT 'extra-in-copy' AS kind, name, ord, line FROM after_ddl
+ EXCEPT
+ SELECT 'extra-in-copy', name, ord, line FROM pgtbl_ddl_rt_orig)
+ORDER BY kind, name, ord;
+ kind | name | ord | line 
+------+------+-----+------
+(0 rows)
+
+-- Cleanup.
+DROP SCHEMA pgtbl_ddl_test CASCADE;
+NOTICE:  drop cascades to 20 other objects
+DETAIL:  drop cascades to table basic
+drop cascades to table id_cols
+drop cascades to table id_custom
+drop cascades to table gen_cols
+drop cascades to table storage_cols
+drop cascades to table refd
+drop cascades to table cons
+drop cascades to table idxd
+drop cascades to table par
+drop cascades to table ch
+drop cascades to table attopt
+drop cascades to table parted_range
+drop cascades to table parted_hash
+drop cascades to table rt_log
+drop cascades to table rt
+drop cascades to table stx
+drop cascades to table rls
+drop cascades to table ri_full
+drop cascades to table ri_idx
+drop cascades to table uno
diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule
index 8fa0a6c47fb..9eca64cca3b 100644
--- a/src/test/regress/parallel_schedule
+++ b/src/test/regress/parallel_schedule
@@ -81,7 +81,7 @@ test: create_table_like alter_generic alter_operator misc async dbsize merge mis
 # collate.linux.utf8 and collate.icu.utf8 tests cannot be run in parallel with each other
 # psql depends on create_am
 # amutils depends on geometry, create_index_spgist, hash_index, brin
-test: rules psql psql_crosstab psql_pipeline amutils stats_ext collate.linux.utf8 collate.windows.win1252
+test: rules psql psql_crosstab psql_pipeline amutils stats_ext collate.linux.utf8 collate.windows.win1252 pg_get_table_ddl
 
 # ----------
 # Run these alone so they don't run out of parallel workers
diff --git a/src/test/regress/sql/pg_get_table_ddl.sql b/src/test/regress/sql/pg_get_table_ddl.sql
new file mode 100644
index 00000000000..c7f72bb824c
--- /dev/null
+++ b/src/test/regress/sql/pg_get_table_ddl.sql
@@ -0,0 +1,278 @@
+--
+-- pg_get_table_ddl
+--
+-- All tests pass owner=>false so the ALTER TABLE OWNER TO line is not
+-- emitted, keeping output stable across test runners (which may run under
+-- different role names).
+--
+CREATE SCHEMA pgtbl_ddl_test;
+SET search_path = pgtbl_ddl_test;
+
+-- Basic table with PRIMARY KEY, NOT NULL, DEFAULT, COLLATE.
+CREATE TABLE basic (
+    id int PRIMARY KEY,
+    name text NOT NULL DEFAULT 'anon' COLLATE "C"
+);
+SELECT * FROM pg_get_table_ddl('basic'::regclass, 'owner', 'false');
+
+-- Identity columns (with default and custom sequence options).
+CREATE TABLE id_cols (
+    id_always int GENERATED ALWAYS AS IDENTITY,
+    id_default int GENERATED BY DEFAULT AS IDENTITY
+);
+SELECT * FROM pg_get_table_ddl('id_cols'::regclass, 'owner', 'false');
+
+CREATE TABLE id_custom (
+    v int GENERATED ALWAYS AS IDENTITY (
+        SEQUENCE NAME id_custom_v_seq
+        START WITH 100 INCREMENT BY 5
+        MINVALUE 50 MAXVALUE 1000
+        CACHE 10 CYCLE
+    )
+);
+SELECT * FROM pg_get_table_ddl('id_custom'::regclass, 'owner', 'false');
+
+-- Generated stored column.
+CREATE TABLE gen_cols (
+    cents int,
+    dollars numeric GENERATED ALWAYS AS (cents / 100.0) STORED
+);
+SELECT * FROM pg_get_table_ddl('gen_cols'::regclass, 'owner', 'false');
+
+-- STORAGE and COMPRESSION (only emitted when non-default for the type).
+CREATE TABLE storage_cols (
+    a text STORAGE EXTERNAL,
+    b text STORAGE MAIN,
+    c text COMPRESSION pglz
+);
+SELECT * FROM pg_get_table_ddl('storage_cols'::regclass, 'owner', 'false');
+
+-- Constraints: CHECK, UNIQUE, FOREIGN KEY (DEFERRABLE).
+CREATE TABLE refd (id int PRIMARY KEY);
+CREATE TABLE cons (
+    a int CHECK (a > 0),
+    b int UNIQUE,
+    c int REFERENCES refd(id) DEFERRABLE INITIALLY DEFERRED
+);
+SELECT * FROM pg_get_table_ddl('cons'::regclass, 'owner', 'false');
+
+-- Indexes: functional and partial.  Constraint-backing indexes are
+-- suppressed (they are emitted by the constraint loop).
+CREATE TABLE idxd (id int PRIMARY KEY, name text);
+CREATE INDEX idxd_lower ON idxd (lower(name));
+CREATE INDEX idxd_partial ON idxd (id) WHERE id > 100;
+SELECT * FROM pg_get_table_ddl('idxd'::regclass, 'owner', 'false');
+
+-- Inheritance, including a child DEFAULT override on an inherited column.
+CREATE TABLE par (a int DEFAULT 1, b text);
+CREATE TABLE ch (c int) INHERITS (par);
+ALTER TABLE ch ALTER COLUMN a SET DEFAULT 999;
+SELECT * FROM pg_get_table_ddl('ch'::regclass, 'owner', 'false');
+
+-- Per-column attoptions: emitted as ALTER COLUMN SET (...).
+CREATE TABLE attopt (a int, b text);
+ALTER TABLE attopt ALTER COLUMN a SET (n_distinct = 100);
+SELECT * FROM pg_get_table_ddl('attopt'::regclass, 'owner', 'false');
+
+-- Partitioned table parent (RANGE and HASH).
+CREATE TABLE parted_range (id int, k int) PARTITION BY RANGE (id);
+SELECT * FROM pg_get_table_ddl('parted_range'::regclass, 'owner', 'false');
+
+CREATE TABLE parted_hash (id int) PARTITION BY HASH (id);
+SELECT * FROM pg_get_table_ddl('parted_hash'::regclass, 'owner', 'false');
+
+-- Partition children: FROM/TO, WITH (modulus, remainder), DEFAULT.
+CREATE TABLE parted_range_1 PARTITION OF parted_range
+    FOR VALUES FROM (0) TO (100);
+ALTER TABLE parted_range_1 ALTER COLUMN k SET DEFAULT 7;
+SELECT * FROM pg_get_table_ddl('parted_range_1'::regclass, 'owner', 'false');
+
+CREATE TABLE parted_hash_0 PARTITION OF parted_hash
+    FOR VALUES WITH (modulus 2, remainder 0);
+SELECT * FROM pg_get_table_ddl('parted_hash_0'::regclass, 'owner', 'false');
+
+CREATE TABLE parted_range_def PARTITION OF parted_range DEFAULT;
+SELECT * FROM pg_get_table_ddl('parted_range_def'::regclass, 'owner', 'false');
+
+-- Rules.
+CREATE TABLE rt (id int);
+CREATE TABLE rt_log (id int);
+CREATE RULE rt_log_insert AS ON INSERT TO rt
+    DO ALSO INSERT INTO rt_log VALUES (NEW.id);
+SELECT * FROM pg_get_table_ddl('rt'::regclass, 'owner', 'false');
+
+-- Extended statistics.
+CREATE TABLE stx (a int, b int, c int);
+CREATE STATISTICS stx_ndv (ndistinct) ON a, b FROM stx;
+SELECT * FROM pg_get_table_ddl('stx'::regclass, 'owner', 'false');
+
+-- Row-level security toggles.
+CREATE TABLE rls (id int);
+ALTER TABLE rls ENABLE ROW LEVEL SECURITY;
+ALTER TABLE rls FORCE ROW LEVEL SECURITY;
+SELECT * FROM pg_get_table_ddl('rls'::regclass, 'owner', 'false');
+
+-- REPLICA IDENTITY: emitted only when not the default.
+CREATE TABLE ri_full (a int);
+ALTER TABLE ri_full REPLICA IDENTITY FULL;
+SELECT * FROM pg_get_table_ddl('ri_full'::regclass, 'owner', 'false');
+
+CREATE TABLE ri_idx (a int NOT NULL);
+CREATE UNIQUE INDEX ri_idx_a ON ri_idx (a);
+ALTER TABLE ri_idx REPLICA IDENTITY USING INDEX ri_idx_a;
+SELECT * FROM pg_get_table_ddl('ri_idx'::regclass, 'owner', 'false');
+
+-- UNLOGGED + reloptions.
+CREATE UNLOGGED TABLE uno (id int) WITH (fillfactor = 70);
+SELECT * FROM pg_get_table_ddl('uno'::regclass, 'owner', 'false');
+
+-- Typed table (CREATE TABLE OF type_name).  Columns inherited from the
+-- type emit nothing; locally-applied DEFAULT, NOT NULL and CHECK come
+-- out through a single "(col WITH OPTIONS ...)" list.
+CREATE TYPE typed_t AS (a int, b text);
+CREATE TABLE typed_plain OF typed_t;
+SELECT * FROM pg_get_table_ddl('typed_plain'::regclass, 'owner', 'false');
+
+CREATE TABLE typed_over OF typed_t (
+    a WITH OPTIONS DEFAULT 7 NOT NULL,
+    b WITH OPTIONS NOT NULL,
+    CONSTRAINT b_nonempty CHECK (length(b) > 0)
+);
+SELECT * FROM pg_get_table_ddl('typed_over'::regclass, 'owner', 'false');
+
+-- Temporary tables + ON COMMIT.  Temp tables live in a session-local
+-- namespace whose name varies between runs, so we strip the schema
+-- prefix.  ON COMMIT DROP only fires at commit, so the queries that
+-- need to see the catalog entry must run in the same transaction as
+-- the CREATE.
+BEGIN;
+CREATE TEMP TABLE temp_default (id int);
+CREATE TEMP TABLE temp_delete (id int) ON COMMIT DELETE ROWS;
+CREATE TEMP TABLE temp_drop (id int) ON COMMIT DROP;
+
+SELECT regexp_replace(line, '"?pg_temp_?[0-9]+"?\.', 'pg_temp.') AS line
+FROM pg_get_table_ddl('temp_default'::regclass, 'owner', 'false') AS line
+WHERE line LIKE 'CREATE %';
+
+SELECT regexp_replace(line, '"?pg_temp_?[0-9]+"?\.', 'pg_temp.') AS line
+FROM pg_get_table_ddl('temp_delete'::regclass, 'owner', 'false') AS line
+WHERE line LIKE 'CREATE %';
+
+SELECT regexp_replace(line, '"?pg_temp_?[0-9]+"?\.', 'pg_temp.') AS line
+FROM pg_get_table_ddl('temp_drop'::regclass, 'owner', 'false') AS line
+WHERE line LIKE 'CREATE %';
+ROLLBACK;
+
+-- Pretty mode.
+SELECT * FROM pg_get_table_ddl('basic'::regclass, 'owner', 'false', 'pretty', 'true');
+
+-- includes_* gating: each sub-object category can be suppressed individually.
+-- includes_indexes=false hides the CREATE INDEX statements.
+SELECT * FROM pg_get_table_ddl('idxd'::regclass, 'owner', 'false',
+                               'includes_indexes', 'false');
+
+-- includes_constraints=false hides both inline CHECK in CREATE TABLE and
+-- the ALTER TABLE ... ADD CONSTRAINT lines.
+SELECT * FROM pg_get_table_ddl('cons'::regclass, 'owner', 'false',
+                               'includes_constraints', 'false');
+
+-- includes_rules=false hides the CREATE RULE.
+SELECT * FROM pg_get_table_ddl('rt'::regclass, 'owner', 'false',
+                               'includes_rules', 'false');
+
+-- includes_statistics=false hides the CREATE STATISTICS.
+SELECT * FROM pg_get_table_ddl('stx'::regclass, 'owner', 'false',
+                               'includes_statistics', 'false');
+
+-- includes_rls=false hides the ENABLE/FORCE ROW LEVEL SECURITY toggles.
+SELECT * FROM pg_get_table_ddl('rls'::regclass, 'owner', 'false',
+                               'includes_rls', 'false');
+
+-- includes_replica_identity=false hides the REPLICA IDENTITY clause.
+SELECT * FROM pg_get_table_ddl('ri_full'::regclass, 'owner', 'false',
+                               'includes_replica_identity', 'false');
+
+-- includes_partition: default is false, so the partitioned-table parent
+-- DDL on its own does not include its children.  Setting it to true
+-- appends the children's DDL after the parent.
+SELECT * FROM pg_get_table_ddl('parted_range'::regclass, 'owner', 'false',
+                               'includes_partition', 'true');
+
+-- Error: not an ordinary or partitioned table.
+CREATE VIEW v AS SELECT 1 AS x;
+SELECT * FROM pg_get_table_ddl('v'::regclass);
+
+CREATE SEQUENCE s;
+SELECT * FROM pg_get_table_ddl('s'::regclass);
+
+-- NULL argument returns no rows.
+SELECT * FROM pg_get_table_ddl(NULL);
+
+-- Unrecognized option.
+SELECT * FROM pg_get_table_ddl('basic'::regclass, 'bogus', 'true');
+
+-- Odd number of variadic args.
+SELECT * FROM pg_get_table_ddl('basic'::regclass, 'pretty');
+
+-- Round-trip verification.  For every test table, capture the DDL the
+-- function emits, drop the schema, replay each table's DDL in
+-- dependency order, and confirm that pg_get_table_ddl on the recreated
+-- relation matches the original line-for-line.  The final SELECT must
+-- return zero rows.
+CREATE TEMP TABLE pgtbl_ddl_rt_orig (name text, ord int, line text);
+INSERT INTO pgtbl_ddl_rt_orig
+SELECT t.name, o.ord, o.line
+FROM (VALUES
+        ('basic'), ('id_cols'), ('id_custom'), ('gen_cols'), ('storage_cols'),
+        ('refd'), ('cons'), ('idxd'),
+        ('par'), ('ch'), ('attopt'),
+        ('parted_range'), ('parted_range_1'), ('parted_range_def'),
+        ('parted_hash'), ('parted_hash_0'),
+        ('rt_log'), ('rt'),
+        ('stx'), ('rls'), ('ri_full'), ('ri_idx'), ('uno')
+     ) AS t(name),
+     LATERAL pg_get_table_ddl(('pgtbl_ddl_test.' || t.name)::regclass,
+                              'owner', 'false') WITH ORDINALITY o(line, ord);
+
+DO $$
+DECLARE
+    tables CONSTANT text[] := ARRAY[
+        'basic', 'id_cols', 'id_custom', 'gen_cols', 'storage_cols',
+        'refd', 'cons', 'idxd',
+        'par', 'ch', 'attopt',
+        'parted_range', 'parted_range_1', 'parted_range_def',
+        'parted_hash', 'parted_hash_0',
+        'rt_log', 'rt',
+        'stx', 'rls', 'ri_full', 'ri_idx', 'uno'
+    ];
+    t text;
+    stmt text;
+BEGIN
+    DROP SCHEMA pgtbl_ddl_test CASCADE;
+    CREATE SCHEMA pgtbl_ddl_test;
+    FOREACH t IN ARRAY tables LOOP
+        FOR stmt IN
+            SELECT line FROM pgtbl_ddl_rt_orig WHERE name = t ORDER BY ord
+        LOOP
+            EXECUTE stmt;
+        END LOOP;
+    END LOOP;
+END $$;
+
+WITH after_ddl AS (
+    SELECT t.name, o.ord, o.line
+    FROM (SELECT DISTINCT name FROM pgtbl_ddl_rt_orig) AS t,
+         LATERAL pg_get_table_ddl(('pgtbl_ddl_test.' || t.name)::regclass,
+                                  'owner', 'false') WITH ORDINALITY o(line, ord)
+)
+(SELECT 'missing-in-copy' AS kind, name, ord, line FROM pgtbl_ddl_rt_orig
+ EXCEPT
+ SELECT 'missing-in-copy', name, ord, line FROM after_ddl)
+UNION ALL
+(SELECT 'extra-in-copy' AS kind, name, ord, line FROM after_ddl
+ EXCEPT
+ SELECT 'extra-in-copy', name, ord, line FROM pgtbl_ddl_rt_orig)
+ORDER BY kind, name, ord;
+
+-- Cleanup.
+DROP SCHEMA pgtbl_ddl_test CASCADE;
-- 
2.51.0