From dc90c22ba99dd48c27e0223954215a326ee7b34c 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 v5] 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>
Reviewed-by: Marcos Pegoraro <marcos@f10.com.br>
Reviewed-by: Zsolt Parragi <zsolt.parragi@percona.com>
---
 doc/src/sgml/func/func-info.sgml              |   79 +
 src/backend/commands/tablecmds.c              |   27 +
 src/backend/utils/adt/ddlutils.c              | 1830 +++++++++++++++++
 src/include/catalog/pg_proc.dat               |    7 +
 src/include/commands/tablecmds.h              |    1 +
 .../regress/expected/pg_get_table_ddl.out     |  742 +++++++
 src/test/regress/parallel_schedule            |    2 +-
 src/test/regress/sql/pg_get_table_ddl.sql     |  440 ++++
 8 files changed, 3127 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..6a34ded9628 100644
--- a/doc/src/sgml/func/func-info.sgml
+++ b/doc/src/sgml/func/func-info.sgml
@@ -3961,6 +3961,85 @@ 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>
+        The <literal>schema_qualified</literal> option (boolean, default
+        <literal>true</literal>) controls whether the target table's own
+        schema is included in the generated DDL.  When set to
+        <literal>false</literal>, the table name is emitted unqualified
+        in the <command>CREATE TABLE</command> and every subsequent
+        <command>ALTER TABLE</command>, <command>CREATE INDEX</command>,
+        <command>CREATE RULE</command>, and
+        <command>CREATE STATISTICS</command> statement.  References to
+        objects in the same schema as the target table (inheritance
+        parents, partition parents, identity sequences, and any
+        same-schema object the deparse helpers happen to mention) are
+        also emitted unqualified, so the script can be replayed under a
+        different <varname>search_path</varname> to recreate the table
+        in another schema.  Cross-schema references (for example a
+        foreign key target in a different schema) remain qualified for
+        correctness.
+       </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 a33e22e8e61..6aa401d943c 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -19558,6 +19558,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..54abc7249c1 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"
@@ -87,6 +102,105 @@ static Datum pg_get_tablespace_ddl_srf(FunctionCallInfo fcinfo, Oid tsid, bool i
 static List *pg_get_database_ddl_internal(Oid dbid, bool pretty,
 										  bool no_owner, bool no_tablespace);
 
+/*
+ * Per-column cache of locally-declared NOT NULL constraints.  Built once
+ * by collect_local_not_null() and consulted by the column-emit helpers
+ * and the post-CREATE constraint loop.  Entries with conoid == InvalidOid
+ * mean the column has no local NOT NULL constraint row in pg_constraint
+ * (the column may still have attnotnull=true on a pre-PG-18-upgraded
+ * catalog, in which case emit plain inline NOT NULL).
+ */
+typedef struct LocalNotNullEntry
+{
+	Oid			conoid;
+	char	   *name;
+	bool		is_auto;		/* matches "<table>_<col>_not_null" */
+	bool		no_inherit;
+}			LocalNotNullEntry;
+
+/*
+ * Working context threaded through the per-pass helpers below.  Inputs
+ * (caller-provided option flags) are filled in once at the top of
+ * pg_get_table_ddl_internal and treated as read-only thereafter.  Derived
+ * fields (qualname, schema_prefix, nn_entries, skip_notnull_oids) are
+ * computed once during setup.  Each pass appends to ctx->buf and pushes
+ * the finished statement onto ctx->statements via append_stmt().
+ */
+typedef struct TableDdlContext
+{
+	Relation	rel;
+	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;
+	bool		schema_qualified;
+
+	/* Derived during setup */
+	Oid			base_namespace;
+	char	   *qualname;
+	char	   *schema_prefix;	/* NULL when schema_qualified */
+	int			save_nestlevel; /* >= 0 if we narrowed search_path */
+	LocalNotNullEntry *nn_entries;
+	List	   *skip_notnull_oids;
+
+	/* Mutable working state */
+	StringInfoData buf;
+	List	   *statements;
+}			TableDdlContext;
+
+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,
+									   bool schema_qualified);
+static void append_column_defs(StringInfo buf, Relation rel, bool pretty,
+							   bool include_constraints,
+							   bool schema_qualified,
+							   LocalNotNullEntry * nn_entries);
+static void append_typed_column_overrides(StringInfo buf, Relation rel,
+										  bool pretty, bool include_constraints,
+										  LocalNotNullEntry * nn_entries);
+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 char *lookup_relname_for_emit(Oid relid, bool schema_qualified,
+									 Oid base_namespace);
+static char *strip_schema_prefix(const char *stmt, const char *prefix);
+static LocalNotNullEntry * collect_local_not_null(Relation rel,
+												  List **skip_oids);
+static List *get_inheritance_parents(Oid relid);
+
+static void append_stmt(TableDdlContext * ctx);
+static void emit_create_table_stmt(TableDdlContext * ctx);
+static void emit_owner_stmt(TableDdlContext * ctx);
+static void emit_child_default_overrides(TableDdlContext * ctx);
+static void emit_attoptions(TableDdlContext * ctx);
+static void emit_indexes(TableDdlContext * ctx);
+static void emit_local_constraints(TableDdlContext * ctx);
+static void emit_rules(TableDdlContext * ctx);
+static void emit_statistics(TableDdlContext * ctx);
+static void emit_replica_identity(TableDdlContext * ctx);
+static void emit_rls_toggles(TableDdlContext * ctx);
+static void emit_partition_children(TableDdlContext * ctx);
+
 
 /*
  * parse_ddl_options
@@ -1185,3 +1299,1719 @@ 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;
+}
+
+/*
+ * lookup_relname_for_emit
+ *		Return either the schema-qualified or the bare quoted name of a
+ *		relation, depending on the schema_qualified flag.
+ *
+ * When schema_qualified is true this is identical to
+ * lookup_qualified_relname.  When false, the bare relname is returned
+ * only if the target relation lives in base_namespace (the namespace
+ * of the table whose DDL is being generated); otherwise the schema-
+ * qualified form is still returned, because cross-schema references
+ * (for example an inheritance parent or foreign key target in a
+ * different schema) are not safe to strip.
+ *
+ * Caller is responsible for pfree()ing the result.
+ */
+static char *
+lookup_relname_for_emit(Oid relid, bool schema_qualified, Oid base_namespace)
+{
+	HeapTuple	tp;
+	Form_pg_class reltup;
+	char	   *result;
+
+	if (schema_qualified)
+		return lookup_qualified_relname(relid);
+
+	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);
+	if (reltup->relnamespace != base_namespace)
+	{
+		/* Cross-schema reference: qualification is required for correctness. */
+		ReleaseSysCache(tp);
+		return lookup_qualified_relname(relid);
+	}
+
+	result = pstrdup(quote_identifier(NameStr(reltup->relname)));
+	ReleaseSysCache(tp);
+
+	return result;
+}
+
+/*
+ * strip_schema_prefix
+ *		Return a palloc'd copy of stmt with every occurrence of prefix
+ *		removed, except occurrences that fall inside a single-quoted SQL
+ *		string literal or a double-quoted SQL identifier.
+ *
+ * Used to strip the qualified-schema prefix (for example "old_schema.")
+ * from DDL emitted by the always-qualified ruleutils helpers
+ * (pg_get_constraintdef_command, pg_get_indexdef_string,
+ * pg_get_ruledef, pg_get_statisticsobjdef_string) when the caller has
+ * requested unqualified output via schema_qualified=false.  A plain
+ * substring strip would corrupt string literals whose contents happen
+ * to contain "<schema>." (for example a CHECK comparing against
+ * 'myschema.secret') or double-quoted column/identifier names whose
+ * contents happen to contain "<schema>." (for example a column named
+ * "s.weird" when stripping the prefix "s."), so we track quoting state
+ * and skip matches that occur while we are inside either kind of
+ * quoted token.  Escaped quotes ('' inside a string, "" inside an
+ * identifier) are handled by keeping the corresponding flag set across
+ * them.
+ */
+static char *
+strip_schema_prefix(const char *stmt, const char *prefix)
+{
+	StringInfoData buf;
+	size_t		plen = strlen(prefix);
+	const char *p = stmt;
+	bool		in_string = false;
+	bool		in_ident = false;
+
+	initStringInfo(&buf);
+	while (*p)
+	{
+		/*
+		 * Try the prefix match BEFORE the quote-state toggles.  The
+		 * prefix itself may begin with a double quote when the base
+		 * schema name requires quoting (e.g. "My-Schema".), and at that
+		 * leading " we must strip the whole prefix rather than treat it
+		 * as the opening of a regular quoted identifier.  The check is
+		 * gated on !in_ident && !in_string, so once we have toggled
+		 * into a quoted token we never strip a coincidental substring
+		 * inside it.
+		 */
+		if (!in_string && !in_ident && strncmp(p, prefix, plen) == 0)
+		{
+			p += plen;
+			continue;
+		}
+		if (*p == '\'' && !in_ident)
+		{
+			/* SQL '' escape: emit both quotes and stay inside the string. */
+			if (in_string && p[1] == '\'')
+			{
+				appendStringInfoChar(&buf, '\'');
+				appendStringInfoChar(&buf, '\'');
+				p += 2;
+				continue;
+			}
+			in_string = !in_string;
+			appendStringInfoChar(&buf, *p);
+			p++;
+			continue;
+		}
+		if (*p == '"' && !in_string)
+		{
+			/* SQL "" escape: emit both quotes and stay inside the ident. */
+			if (in_ident && p[1] == '"')
+			{
+				appendStringInfoChar(&buf, '"');
+				appendStringInfoChar(&buf, '"');
+				p += 2;
+				continue;
+			}
+			in_ident = !in_ident;
+			appendStringInfoChar(&buf, *p);
+			p++;
+			continue;
+		}
+		appendStringInfoChar(&buf, *p);
+		p++;
+	}
+	return buf.data;
+}
+
+/*
+ * collect_local_not_null
+ *		Scan pg_constraint once for locally-declared NOT NULL constraints
+ *		on rel, returning a palloc'd array indexed by attnum (1..natts).
+ *
+ * Entries with conoid==InvalidOid mean "no local NOT NULL row for this
+ * column".  When the constraint name does not match the auto-generated
+ * pattern "<tablename>_<columnname>_not_null", the constraint OID is
+ * appended to *skip_oids so the post-CREATE constraint loop can avoid
+ * re-emitting it as ALTER TABLE … ADD CONSTRAINT — the column-emit
+ * pass will produce it inline as "CONSTRAINT name NOT NULL" instead.
+ */
+static LocalNotNullEntry *
+collect_local_not_null(Relation rel, List **skip_oids)
+{
+	TupleDesc	tupdesc = RelationGetDescr(rel);
+	int			natts = tupdesc->natts;
+	LocalNotNullEntry *entries;
+	const char *relname = RelationGetRelationName(rel);
+	Relation	conRel;
+	SysScanDesc conScan;
+	ScanKeyData conKey;
+	HeapTuple	conTup;
+
+	entries = (LocalNotNullEntry *) palloc0(sizeof(LocalNotNullEntry) *
+											(natts + 1));
+
+	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		conkeyDat;
+		bool		conkeyNull;
+		ArrayType  *conkeyArr;
+		int16	   *conkeyVals;
+		int			attnum;
+		Form_pg_attribute att;
+		char		autoname[NAMEDATALEN];
+
+		if (con->contype != CONSTRAINT_NOTNULL)
+			continue;
+		if (!con->conislocal)
+			continue;
+
+		conkeyDat = heap_getattr(conTup, Anum_pg_constraint_conkey,
+								 RelationGetDescr(conRel), &conkeyNull);
+		if (conkeyNull)
+			continue;
+		conkeyArr = DatumGetArrayTypeP(conkeyDat);
+
+		/*
+		 * Defend against a malformed conkey: a NOT NULL constraint is always
+		 * a single-column 1-D int2 array, but a corrupted catalog or a future
+		 * patch that stores wider conkeys mustn't trip us into reading past
+		 * the array header.
+		 */
+		if (ARR_NDIM(conkeyArr) != 1 ||
+			ARR_DIMS(conkeyArr)[0] < 1 ||
+			ARR_HASNULL(conkeyArr) ||
+			ARR_ELEMTYPE(conkeyArr) != INT2OID)
+			continue;
+
+		conkeyVals = (int16 *) ARR_DATA_PTR(conkeyArr);
+		attnum = conkeyVals[0];
+		if (attnum < 1 || attnum > natts)
+			continue;
+
+		att = TupleDescAttr(tupdesc, attnum - 1);
+		snprintf(autoname, sizeof(autoname), "%s_%s_not_null",
+				 relname, NameStr(att->attname));
+
+		entries[attnum].conoid = con->oid;
+		entries[attnum].name = pstrdup(NameStr(con->conname));
+		entries[attnum].is_auto = (strcmp(NameStr(con->conname), autoname) == 0);
+		entries[attnum].no_inherit = con->connoinherit;
+
+		/*
+		 * Inline emission of "CONSTRAINT name NOT NULL" only happens for
+		 * columns that the column list actually emits, i.e. attislocal
+		 * columns.  For a locally-declared NOT NULL sitting on an inherited
+		 * column the inline path never fires, so we must leave the constraint
+		 * OID out of skip_oids and let the post-CREATE constraint loop emit
+		 * ALTER TABLE ... ADD CONSTRAINT.
+		 */
+		if (!entries[attnum].is_auto && att->attislocal && skip_oids != NULL)
+			*skip_oids = lappend_oid(*skip_oids, con->oid);
+	}
+	systable_endscan(conScan);
+	table_close(conRel, AccessShareLock);
+
+	return entries;
+}
+
+/*
+ * 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,
+				   bool schema_qualified,
+				   LocalNotNullEntry * nn_entries)
+{
+	TupleDesc	tupdesc = RelationGetDescr(rel);
+	Oid			base_namespace = RelationGetNamespace(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_relname_for_emit(seqid,
+															schema_qualified,
+															base_namespace);
+
+								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)
+		{
+			LocalNotNullEntry *nn = &nn_entries[att->attnum];
+
+			if (nn->name != NULL && !nn->is_auto)
+				appendStringInfo(buf, " CONSTRAINT %s NOT NULL",
+								 quote_identifier(nn->name));
+			else
+				appendStringInfoString(buf, " NOT NULL");
+			if (nn->name != NULL && nn->no_inherit)
+				appendStringInfoString(buf, " NO INHERIT");
+		}
+	}
+
+	/*
+	 * 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,
+							  LocalNotNullEntry * nn_entries)
+{
+	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)
+		{
+			LocalNotNullEntry *nn = &nn_entries[att->attnum];
+
+			if (nn->name != NULL && !nn->is_auto)
+				appendStringInfo(&inner, " CONSTRAINT %s NOT NULL",
+								 quote_identifier(nn->name));
+			else
+				appendStringInfoString(&inner, " NOT NULL");
+			if (nn->name != NULL && nn->no_inherit)
+				appendStringInfoString(&inner, " NO INHERIT");
+		}
+
+		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);
+}
+
+/*
+ * append_stmt / append_stripped_stmt
+ *		Push ctx->buf onto ctx->statements.
+ *
+ * append_stmt is used for emissions where the table reference is built
+ * from ctx->qualname (already partial-qualified per schema_qualified).
+ * append_stripped_stmt is used after invoking always-qualified ruleutils
+ * helpers (pg_get_constraintdef_command, pg_get_indexdef_string,
+ * pg_get_ruledef, pg_get_statisticsobjdef_string) — when
+ * schema_qualified is false, the base-schema prefix is stripped from
+ * their output so the emitted DDL is replay-relocatable.
+ */
+static void
+append_stmt(TableDdlContext * ctx)
+{
+	ctx->statements = lappend(ctx->statements, pstrdup(ctx->buf.data));
+}
+
+static void
+append_stripped_stmt(TableDdlContext * ctx)
+{
+	if (ctx->schema_prefix != NULL)
+		ctx->statements = lappend(ctx->statements,
+								  strip_schema_prefix(ctx->buf.data,
+													  ctx->schema_prefix));
+	else
+		ctx->statements = lappend(ctx->statements, pstrdup(ctx->buf.data));
+}
+
+/*
+ * emit_create_table_stmt
+ *		Build the leading CREATE TABLE statement, including persistence
+ *		(TEMPORARY / UNLOGGED), body (column list / OF type_name /
+ *		PARTITION OF parent), INHERITS, PARTITION BY, USING method,
+ *		WITH (reloptions), TABLESPACE, and ON COMMIT.
+ */
+static void
+emit_create_table_stmt(TableDdlContext * ctx)
+{
+	Relation	rel = ctx->rel;
+	char		relpersistence = rel->rd_rel->relpersistence;
+	char		relkind = rel->rd_rel->relkind;
+	bool		is_typed = OidIsValid(rel->rd_rel->reloftype);
+	HeapTuple	classtup;
+	Datum		reloptDatum;
+	bool		reloptIsnull;
+
+	classtup = SearchSysCache1(RELOID, ObjectIdGetDatum(ctx->relid));
+	if (!HeapTupleIsValid(classtup))
+		elog(ERROR, "cache lookup failed for relation %u", ctx->relid);
+
+	reloptDatum = SysCacheGetAttr(RELOID, classtup,
+								  Anum_pg_class_reloptions, &reloptIsnull);
+
+	resetStringInfo(&ctx->buf);
+	appendStringInfoString(&ctx->buf, "CREATE ");
+	if (relpersistence == RELPERSISTENCE_TEMP)
+		appendStringInfoString(&ctx->buf, "TEMPORARY ");
+	else if (relpersistence == RELPERSISTENCE_UNLOGGED)
+		appendStringInfoString(&ctx->buf, "UNLOGGED ");
+	appendStringInfo(&ctx->buf, "TABLE %s", ctx->qualname);
+
+	if (rel->rd_rel->relispartition)
+	{
+		Oid			parentOid = get_partition_parent(ctx->relid, true);
+		char	   *parentQual = lookup_relname_for_emit(parentOid,
+														 ctx->schema_qualified,
+														 ctx->base_namespace);
+		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(&ctx->buf, " PARTITION OF %s %s",
+						 parentQual, forValues ? forValues : "DEFAULT");
+		if (forValues)
+			pfree(forValues);
+		if (boundStr)
+			pfree(boundStr);
+		pfree(parentQual);
+		pfree(parentRelname);
+	}
+	else if (is_typed)
+	{
+		char	   *typname = format_type_be_qualified(rel->rd_rel->reloftype);
+
+		appendStringInfo(&ctx->buf, " OF %s", typname);
+		pfree(typname);
+
+		append_typed_column_overrides(&ctx->buf, rel, ctx->pretty,
+									  ctx->include_constraints,
+									  ctx->nn_entries);
+	}
+	else
+	{
+		List	   *parents;
+		ListCell   *lc;
+		bool		first;
+
+		appendStringInfoString(&ctx->buf, " (");
+
+		append_column_defs(&ctx->buf, rel, ctx->pretty,
+						   ctx->include_constraints,
+						   ctx->schema_qualified, ctx->nn_entries);
+
+		if (ctx->pretty)
+			appendStringInfoString(&ctx->buf, "\n)");
+		else
+			appendStringInfoChar(&ctx->buf, ')');
+
+		parents = get_inheritance_parents(ctx->relid);
+		if (parents != NIL)
+		{
+			appendStringInfoString(&ctx->buf, " INHERITS (");
+			first = true;
+			foreach(lc, parents)
+			{
+				Oid			poid = lfirst_oid(lc);
+				char	   *pname = lookup_relname_for_emit(poid,
+															ctx->schema_qualified,
+															ctx->base_namespace);
+
+				if (!first)
+					appendStringInfoString(&ctx->buf, ", ");
+				first = false;
+				appendStringInfoString(&ctx->buf, pname);
+				pfree(pname);
+			}
+			appendStringInfoChar(&ctx->buf, ')');
+			list_free(parents);
+		}
+	}
+
+	if (relkind == RELKIND_PARTITIONED_TABLE)
+	{
+		Datum		partkeyDatum;
+		char	   *partkey;
+
+		partkeyDatum = OidFunctionCall1(F_PG_GET_PARTKEYDEF,
+										ObjectIdGetDatum(ctx->relid));
+		partkey = TextDatumGetCString(partkeyDatum);
+		appendStringInfo(&ctx->buf, " PARTITION BY %s", partkey);
+		pfree(partkey);
+	}
+
+	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(&ctx->buf, " USING %s",
+							 quote_identifier(amname));
+			pfree(amname);
+		}
+	}
+
+	if (!reloptIsnull)
+	{
+		appendStringInfoString(&ctx->buf, " WITH (");
+		get_reloptions(&ctx->buf, reloptDatum);
+		appendStringInfoChar(&ctx->buf, ')');
+	}
+
+	ReleaseSysCache(classtup);
+
+	if (!ctx->no_tablespace && OidIsValid(rel->rd_rel->reltablespace))
+	{
+		char	   *tsname = get_tablespace_name(rel->rd_rel->reltablespace);
+
+		if (tsname != NULL)
+		{
+			appendStringInfo(&ctx->buf, " TABLESPACE %s",
+							 quote_identifier(tsname));
+			pfree(tsname);
+		}
+	}
+
+	if (relpersistence == RELPERSISTENCE_TEMP)
+	{
+		OnCommitAction oc = get_on_commit_action(ctx->relid);
+
+		if (oc == ONCOMMIT_DELETE_ROWS)
+			appendStringInfoString(&ctx->buf, " ON COMMIT DELETE ROWS");
+		else if (oc == ONCOMMIT_DROP)
+			appendStringInfoString(&ctx->buf, " ON COMMIT DROP");
+	}
+
+	appendStringInfoChar(&ctx->buf, ';');
+	append_stmt(ctx);
+}
+
+/*
+ * emit_owner_stmt
+ *		ALTER TABLE qualname OWNER TO role.
+ */
+static void
+emit_owner_stmt(TableDdlContext * ctx)
+{
+	char	   *owner;
+
+	if (ctx->no_owner)
+		return;
+
+	owner = GetUserNameFromId(ctx->rel->rd_rel->relowner, false);
+	resetStringInfo(&ctx->buf);
+	appendStringInfo(&ctx->buf, "ALTER TABLE %s OWNER TO %s;",
+					 ctx->qualname, quote_identifier(owner));
+	append_stmt(ctx);
+	pfree(owner);
+}
+
+/*
+ * emit_child_default_overrides
+ *		ALTER TABLE qualname ALTER COLUMN col SET DEFAULT expr — one per
+ *		inherited (attislocal=false) non-generated column carrying a
+ *		locally-set default.  Generated columns are skipped: their
+ *		expression is inherited automatically and SET DEFAULT would
+ *		fail at replay.
+ */
+static void
+emit_child_default_overrides(TableDdlContext * ctx)
+{
+	TupleDesc	tupdesc = RelationGetDescr(ctx->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;
+		if (att->attgenerated != '\0')
+			continue;
+
+		defstr = find_attrdef_text(ctx->rel, att->attnum, &dpcontext);
+		if (defstr == NULL)
+			continue;
+
+		resetStringInfo(&ctx->buf);
+		appendStringInfo(&ctx->buf,
+						 "ALTER TABLE %s ALTER COLUMN %s SET DEFAULT %s;",
+						 ctx->qualname,
+						 quote_identifier(NameStr(att->attname)),
+						 defstr);
+		append_stmt(ctx);
+		pfree(defstr);
+	}
+}
+
+/*
+ * emit_attoptions
+ *		ALTER TABLE qualname ALTER COLUMN col SET (...) — one per column
+ *		with non-null pg_attribute.attoptions.  The inline form of these
+ *		options isn't available in CREATE TABLE, so they come out here.
+ */
+static void
+emit_attoptions(TableDdlContext * ctx)
+{
+	TupleDesc	tupdesc = RelationGetDescr(ctx->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(ctx->relid),
+								 Int16GetDatum(att->attnum));
+		if (!HeapTupleIsValid(attTup))
+			continue;
+
+		optDatum = SysCacheGetAttr(ATTNUM, attTup,
+								   Anum_pg_attribute_attoptions, &optIsnull);
+		if (!optIsnull)
+		{
+			resetStringInfo(&ctx->buf);
+			appendStringInfo(&ctx->buf,
+							 "ALTER TABLE %s ALTER COLUMN %s SET (",
+							 ctx->qualname,
+							 quote_identifier(NameStr(att->attname)));
+			get_reloptions(&ctx->buf, optDatum);
+			appendStringInfoString(&ctx->buf, ");");
+			append_stmt(ctx);
+		}
+		ReleaseSysCache(attTup);
+	}
+}
+
+/*
+ * emit_indexes
+ *		CREATE INDEX per non-constraint-backed index on the relation.
+ *		Indexes that back PK / UNIQUE / EXCLUDE constraints are emitted
+ *		out-of-line by emit_local_constraints (the ALTER TABLE … ADD
+ *		CONSTRAINT statement creates the index implicitly).
+ */
+static void
+emit_indexes(TableDdlContext * ctx)
+{
+	List	   *indexoids;
+	ListCell   *lc;
+
+	if (!ctx->include_indexes)
+		return;
+
+	indexoids = RelationGetIndexList(ctx->rel);
+	foreach(lc, indexoids)
+	{
+		Oid			idxoid = lfirst_oid(lc);
+		char	   *idxdef;
+
+		if (OidIsValid(get_index_constraint(idxoid)))
+			continue;
+
+		idxdef = pg_get_indexdef_string(idxoid);
+		resetStringInfo(&ctx->buf);
+		appendStringInfo(&ctx->buf, "%s;", idxdef);
+		append_stripped_stmt(ctx);
+		pfree(idxdef);
+	}
+	list_free(indexoids);
+}
+
+/*
+ * emit_local_constraints
+ *		ALTER TABLE … ADD CONSTRAINT for each locally-defined constraint
+ *		on the relation.  Inherited constraints (conislocal=false) come
+ *		from the parent's DDL.  CHECK constraints are emitted inline for
+ *		regular/typed tables (skip here) but out-of-line for partition
+ *		children (no column list to live in).  User-named NOT NULL
+ *		constraints are emitted inline by the column-emit helpers and
+ *		are skipped via skip_notnull_oids; auto-named NOT NULLs fall
+ *		through here and are dedup'd by name at replay.
+ */
+static void
+emit_local_constraints(TableDdlContext * ctx)
+{
+	Relation	conRel;
+	SysScanDesc conScan;
+	ScanKeyData conKey;
+	HeapTuple	conTup;
+	bool		is_partition = ctx->rel->rd_rel->relispartition;
+
+	if (!ctx->include_constraints)
+		return;
+
+	conRel = table_open(ConstraintRelationId, AccessShareLock);
+	ScanKeyInit(&conKey,
+				Anum_pg_constraint_conrelid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(ctx->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;
+		if (con->contype == CONSTRAINT_CHECK && !is_partition)
+			continue;
+		if (con->contype == CONSTRAINT_NOTNULL && !is_partition &&
+			list_member_oid(ctx->skip_notnull_oids, con->oid))
+			continue;
+
+		condef = pg_get_constraintdef_command(con->oid);
+		resetStringInfo(&ctx->buf);
+		appendStringInfo(&ctx->buf, "%s;", condef);
+		append_stripped_stmt(ctx);
+		pfree(condef);
+	}
+	systable_endscan(conScan);
+	table_close(conRel, AccessShareLock);
+}
+
+/*
+ * emit_rules
+ *		CREATE RULE per cached rewrite rule on the relation.
+ */
+static void
+emit_rules(TableDdlContext * ctx)
+{
+	if (!ctx->include_rules || ctx->rel->rd_rules == NULL)
+		return;
+
+	for (int i = 0; i < ctx->rel->rd_rules->numLocks; i++)
+	{
+		Oid			ruleid = ctx->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(&ctx->buf);
+		appendStringInfoString(&ctx->buf, ruledef_str);
+		append_stripped_stmt(ctx);
+		pfree(ruledef_str);
+	}
+}
+
+/*
+ * emit_statistics
+ *		CREATE STATISTICS per extended statistics object on the relation.
+ */
+static void
+emit_statistics(TableDdlContext * ctx)
+{
+	Relation	statRel;
+	SysScanDesc statScan;
+	ScanKeyData statKey;
+	HeapTuple	statTup;
+
+	if (!ctx->include_statistics)
+		return;
+
+	statRel = table_open(StatisticExtRelationId, AccessShareLock);
+	ScanKeyInit(&statKey,
+				Anum_pg_statistic_ext_stxrelid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(ctx->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(&ctx->buf);
+		appendStringInfo(&ctx->buf, "%s;", statdef);
+		append_stripped_stmt(ctx);
+		pfree(statdef);
+	}
+	systable_endscan(statScan);
+	table_close(statRel, AccessShareLock);
+}
+
+/*
+ * emit_replica_identity
+ *		ALTER TABLE qualname REPLICA IDENTITY … — emitted only when the
+ *		relreplident differs from the default ('d' = use primary key).
+ */
+static void
+emit_replica_identity(TableDdlContext * ctx)
+{
+	if (!ctx->include_replica_identity)
+		return;
+	if (ctx->rel->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT)
+		return;
+
+	resetStringInfo(&ctx->buf);
+	switch (ctx->rel->rd_rel->relreplident)
+	{
+		case REPLICA_IDENTITY_NOTHING:
+			appendStringInfo(&ctx->buf,
+							 "ALTER TABLE %s REPLICA IDENTITY NOTHING;",
+							 ctx->qualname);
+			append_stmt(ctx);
+			break;
+		case REPLICA_IDENTITY_FULL:
+			appendStringInfo(&ctx->buf,
+							 "ALTER TABLE %s REPLICA IDENTITY FULL;",
+							 ctx->qualname);
+			append_stmt(ctx);
+			break;
+		case REPLICA_IDENTITY_INDEX:
+			{
+				Oid			replidx = RelationGetReplicaIndex(ctx->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(&ctx->buf,
+									 "ALTER TABLE %s REPLICA IDENTITY USING INDEX %s;",
+									 ctx->qualname,
+									 quote_identifier(idxname));
+					append_stmt(ctx);
+					pfree(idxname);
+				}
+			}
+			break;
+	}
+}
+
+/*
+ * emit_rls_toggles
+ *		ALTER TABLE qualname ENABLE / FORCE ROW LEVEL SECURITY.
+ */
+static void
+emit_rls_toggles(TableDdlContext * ctx)
+{
+	if (!ctx->include_rls)
+		return;
+
+	if (ctx->rel->rd_rel->relrowsecurity)
+	{
+		resetStringInfo(&ctx->buf);
+		appendStringInfo(&ctx->buf,
+						 "ALTER TABLE %s ENABLE ROW LEVEL SECURITY;",
+						 ctx->qualname);
+		append_stmt(ctx);
+	}
+	if (ctx->rel->rd_rel->relforcerowsecurity)
+	{
+		resetStringInfo(&ctx->buf);
+		appendStringInfo(&ctx->buf,
+						 "ALTER TABLE %s FORCE ROW LEVEL SECURITY;",
+						 ctx->qualname);
+		append_stmt(ctx);
+	}
+}
+
+/*
+ * emit_partition_children
+ *		For each direct partition child of a partitioned-table parent,
+ *		recursively call pg_get_table_ddl_internal and append the child's
+ *		statements.  Each child's own DDL handles further levels through
+ *		the same recursion.
+ */
+static void
+emit_partition_children(TableDdlContext * ctx)
+{
+	List	   *children;
+	ListCell   *lc;
+
+	if (!ctx->include_partition ||
+		ctx->rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)
+		return;
+
+	children = find_inheritance_children(ctx->relid, AccessShareLock);
+	foreach(lc, children)
+	{
+		Oid			childoid = lfirst_oid(lc);
+		List	   *childstmts;
+
+		childstmts = pg_get_table_ddl_internal(childoid, ctx->pretty,
+											   ctx->no_owner,
+											   ctx->no_tablespace,
+											   ctx->include_indexes,
+											   ctx->include_constraints,
+											   ctx->include_rules,
+											   ctx->include_statistics,
+											   ctx->include_triggers,
+											   ctx->include_policies,
+											   ctx->include_rls,
+											   ctx->include_replica_identity,
+											   ctx->include_partition,
+											   ctx->schema_qualified);
+		ctx->statements = list_concat(ctx->statements, childstmts);
+	}
+	list_free(children);
+}
+
+/*
+ * 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 currently disabled
+ * (#if 0) — they will become a single helper call once the standalone
+ * pg_get_trigger_ddl / pg_get_policy_ddl helpers land.
+ */
+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,
+						  bool schema_qualified)
+{
+	TableDdlContext ctx = {0};
+	Relation	rel;
+	char		relkind;
+	AclResult	aclresult;
+
+	rel = table_open(relid, AccessShareLock);
+
+	relkind = rel->rd_rel->relkind;
+
+	/*
+	 * 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));
+
+	ctx.rel = rel;
+	ctx.relid = relid;
+	ctx.pretty = pretty;
+	ctx.no_owner = no_owner;
+	ctx.no_tablespace = no_tablespace;
+	ctx.include_indexes = include_indexes;
+	ctx.include_constraints = include_constraints;
+	ctx.include_rules = include_rules;
+	ctx.include_statistics = include_statistics;
+	ctx.include_triggers = include_triggers;
+	ctx.include_policies = include_policies;
+	ctx.include_rls = include_rls;
+	ctx.include_replica_identity = include_replica_identity;
+	ctx.include_partition = include_partition;
+	ctx.schema_qualified = schema_qualified;
+	ctx.base_namespace = RelationGetNamespace(rel);
+	ctx.save_nestlevel = -1;
+	ctx.qualname = lookup_relname_for_emit(relid, schema_qualified,
+										   ctx.base_namespace);
+
+	/*
+	 * When schema_qualified is false, narrow the active search_path to the
+	 * target table's own namespace for the duration of DDL emission.  The
+	 * deparse code in ruleutils (and the regclass/regproc output functions we
+	 * transitively call) decides whether to schema-qualify a name based on
+	 * whether the referenced object is reachable through search_path, so this
+	 * makes same-schema references in DEFAULT and CHECK expressions, regclass
+	 * literals, FK targets, etc. come out unqualified.  Cross-schema
+	 * references stay qualified, which is the correctness requirement.
+	 * AtEOXact_GUC cleans up at xact end if anything throws between here and
+	 * the explicit restore below; on the normal path we restore right before
+	 * returning.
+	 *
+	 * Pre-compute "<schema>." too — the always-qualified helpers
+	 * (constraintdef_command, indexdef_string, ruledef,
+	 * statisticsobjdef_string) ignore search_path for the target's own "ALTER
+	 * TABLE schema.tbl"/"ON schema.tbl"/"TO schema.tbl" prefix and for the
+	 * qualified statistics name; we strip that string ourselves via
+	 * strip_schema_prefix.
+	 */
+	if (!schema_qualified)
+	{
+		char	   *nspname = get_namespace_name(ctx.base_namespace);
+
+		if (nspname != NULL)
+		{
+			const char *qnsp = quote_identifier(nspname);
+
+			ctx.schema_prefix = psprintf("%s.", qnsp);
+			ctx.save_nestlevel = NewGUCNestLevel();
+			(void) set_config_option("search_path", qnsp,
+									 PGC_USERSET, PGC_S_SESSION,
+									 GUC_ACTION_SAVE, true, 0, false);
+			pfree(nspname);
+		}
+	}
+
+	/*
+	 * Cache locally-declared NOT NULL constraint metadata so the column- emit
+	 * helpers can produce "CONSTRAINT name NOT NULL" inline for user-named
+	 * constraints, and so the constraint loop can avoid double-emitting them.
+	 */
+	ctx.nn_entries = collect_local_not_null(rel, &ctx.skip_notnull_oids);
+
+	initStringInfo(&ctx.buf);
+
+	/*
+	 * Each pass below appends zero or more finished statements to
+	 * ctx.statements via append_stmt / append_stripped_stmt.  Order is
+	 * significant: CREATE TABLE first; OWNER and the per-column ALTER COLUMN
+	 * passes before sub-object emission; sub-objects in dependency-friendly
+	 * order (indexes before constraints, since constraint-backed indexes are
+	 * emitted out-of-line by the constraint loop); and partition children
+	 * last so the parent already exists at replay time.
+	 */
+	emit_create_table_stmt(&ctx);
+	emit_owner_stmt(&ctx);
+	emit_child_default_overrides(&ctx);
+	emit_attoptions(&ctx);
+	emit_indexes(&ctx);
+	emit_local_constraints(&ctx);
+	emit_rules(&ctx);
+	emit_statistics(&ctx);
+	emit_replica_identity(&ctx);
+	emit_rls_toggles(&ctx);
+
+	/*
+	 * Triggers and row-level security policies — disabled until the
+	 * standalone pg_get_trigger_ddl() and pg_get_policy_ddl() helpers land.
+	 * The scan + lock + filter scaffolding below is preserved (inside #if 0)
+	 * so wiring up each emission becomes a one-liner in the loop body once
+	 * those helpers are available.
+	 */
+#if 0
+	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. */
+		}
+		systable_endscan(trigScan);
+		table_close(trigRel, AccessShareLock);
+	}
+
+	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)))
+		{
+			/* TODO: append pg_get_policy_ddl(relid, polname) output here. */
+		}
+		systable_endscan(polScan);
+		table_close(polRel, AccessShareLock);
+	}
+#else
+	(void) include_triggers;
+	(void) include_policies;
+#endif
+
+	emit_partition_children(&ctx);
+
+	{
+		int			natts = RelationGetDescr(rel)->natts;
+
+		for (int i = 1; i <= natts; i++)
+			if (ctx.nn_entries[i].name != NULL)
+				pfree(ctx.nn_entries[i].name);
+	}
+	pfree(ctx.nn_entries);
+
+	table_close(rel, AccessShareLock);
+	pfree(ctx.buf.data);
+	pfree(ctx.qualname);
+	if (ctx.schema_prefix != NULL)
+		pfree(ctx.schema_prefix);
+	list_free(ctx.skip_notnull_oids);
+
+	/*
+	 * Pop the narrowed search_path now that all helpers have run.  Errors
+	 * thrown earlier are cleaned up by AtEOXact_GUC at xact end.
+	 */
+	if (ctx.save_nestlevel >= 0)
+		AtEOXact_GUC(true, ctx.save_nestlevel);
+
+	return ctx.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},
+			{"schema_qualified", 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,
+											   !opts[12].isset || opts[12].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..e456c226f53 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', proparallel => 'r',
+  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..031c4d9f130
--- /dev/null
+++ b/src/test/regress/expected/pg_get_table_ddl.out
@@ -0,0 +1,742 @@
+--
+-- 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)
+
+-- schema_qualified=false strips the schema from the target table itself
+-- and from same-schema sibling references (inheritance/partition
+-- parents, identity sequences in the same schema), while cross-schema
+-- references are kept qualified for correctness.
+--
+-- Bare CREATE TABLE for an ordinary table.
+SELECT * FROM pg_get_table_ddl('basic'::regclass, 'owner', 'false',
+                               'schema_qualified', 'false');
+                                        pg_get_table_ddl                                         
+-------------------------------------------------------------------------------------------------
+ CREATE TABLE basic ( id integer NOT NULL, name text COLLATE "C" DEFAULT 'anon'::text NOT NULL);
+ ALTER TABLE basic ADD CONSTRAINT basic_id_not_null NOT NULL id;
+ ALTER TABLE basic ADD CONSTRAINT basic_name_not_null NOT NULL name;
+ ALTER TABLE basic ADD CONSTRAINT basic_pkey PRIMARY KEY (id);
+(4 rows)
+
+-- INHERITS parent in the same schema is emitted unqualified.
+SELECT * FROM pg_get_table_ddl('ch'::regclass, 'owner', 'false',
+                               'schema_qualified', 'false');
+                pg_get_table_ddl                
+------------------------------------------------
+ CREATE TABLE ch ( c integer) INHERITS (par);
+ ALTER TABLE ch ALTER COLUMN a SET DEFAULT 999;
+(2 rows)
+
+-- Partition parent in the same schema is emitted unqualified.
+SELECT * FROM pg_get_table_ddl('parted_range_1'::regclass, 'owner', 'false',
+                               'schema_qualified', 'false');
+                                  pg_get_table_ddl                                   
+-------------------------------------------------------------------------------------
+ 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;
+(2 rows)
+
+-- Cross-schema FK target keeps qualification.  The 'refd' table is in
+-- pgtbl_ddl_test; place an FK target in a different schema and verify it
+-- stays qualified even with schema_qualified=false.
+CREATE SCHEMA pgtbl_ddl_other;
+CREATE TABLE pgtbl_ddl_other.parent (id int PRIMARY KEY);
+CREATE TABLE xschema_fk (
+    id  int PRIMARY KEY,
+    pid int REFERENCES pgtbl_ddl_other.parent
+);
+SELECT * FROM pg_get_table_ddl('xschema_fk'::regclass, 'owner', 'false',
+                               'schema_qualified', 'false');
+                                                  pg_get_table_ddl                                                  
+--------------------------------------------------------------------------------------------------------------------
+ CREATE TABLE xschema_fk ( id integer NOT NULL, pid integer);
+ ALTER TABLE xschema_fk ADD CONSTRAINT xschema_fk_id_not_null NOT NULL id;
+ ALTER TABLE xschema_fk ADD CONSTRAINT xschema_fk_pid_fkey FOREIGN KEY (pid) REFERENCES pgtbl_ddl_other.parent(id);
+ ALTER TABLE xschema_fk ADD CONSTRAINT xschema_fk_pkey PRIMARY KEY (id);
+(4 rows)
+
+-- Identity column with a custom sequence name in the same schema is
+-- emitted unqualified.
+SELECT * FROM pg_get_table_ddl('id_custom'::regclass, 'owner', 'false',
+                               'schema_qualified', 'false');
+                                                                  pg_get_table_ddl                                                                   
+-----------------------------------------------------------------------------------------------------------------------------------------------------
+ CREATE TABLE 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 id_custom ADD CONSTRAINT id_custom_v_not_null NOT NULL v;
+(2 rows)
+
+-- Round-trip: with schema_qualified=false, after SET search_path the
+-- DDL can be replayed into a different target schema.
+CREATE SCHEMA pgtbl_ddl_replay;
+DO $$
+DECLARE
+    stmt text;
+BEGIN
+    SET LOCAL search_path = pgtbl_ddl_replay;
+    FOR stmt IN
+        SELECT line FROM pg_get_table_ddl('pgtbl_ddl_test.basic'::regclass,
+                                          'owner', 'false',
+                                          'schema_qualified', 'false') AS line
+    LOOP
+        EXECUTE stmt;
+    END LOOP;
+END $$;
+SELECT relnamespace::regnamespace::text, relname
+FROM pg_class WHERE oid = 'pgtbl_ddl_replay.basic'::regclass;
+   relnamespace   | relname 
+------------------+---------
+ pgtbl_ddl_replay | basic
+(1 row)
+
+DROP SCHEMA pgtbl_ddl_replay CASCADE;
+NOTICE:  drop cascades to table pgtbl_ddl_replay.basic
+DROP TABLE xschema_fk;
+DROP SCHEMA pgtbl_ddl_other CASCADE;
+NOTICE:  drop cascades to table pgtbl_ddl_other.parent
+-- Locally-declared CHECK on a partition child has no inline column
+-- list to live in (PARTITION OF form), so it must come out as a
+-- separate ALTER TABLE ... ADD CONSTRAINT statement.
+CREATE TABLE parted_chk (id int, val int) PARTITION BY RANGE (id);
+CREATE TABLE parted_chk_child PARTITION OF parted_chk
+    (CONSTRAINT chk_inline CHECK (val > 0))
+    FOR VALUES FROM (0) TO (100);
+SELECT * FROM pg_get_table_ddl('parted_chk_child'::regclass, 'owner', 'false');
+                                                 pg_get_table_ddl                                                  
+-------------------------------------------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.parted_chk_child PARTITION OF pgtbl_ddl_test.parted_chk FOR VALUES FROM (0) TO (100);
+ ALTER TABLE pgtbl_ddl_test.parted_chk_child ADD CONSTRAINT chk_inline CHECK ((val > 0));
+(2 rows)
+
+DROP TABLE parted_chk;
+-- Inheritance children of a parent with a generated column should not
+-- emit an ALTER COLUMN ... SET DEFAULT for the inherited generated
+-- column: the GENERATED expression carries down through inheritance,
+-- and SET DEFAULT would fail at replay against a generated column.
+CREATE TABLE par_gen (
+    id int,
+    g  int GENERATED ALWAYS AS (id * 2) STORED
+);
+CREATE TABLE ch_gen () INHERITS (par_gen);
+SELECT * FROM pg_get_table_ddl('ch_gen'::regclass, 'owner', 'false');
+                             pg_get_table_ddl                             
+--------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.ch_gen () INHERITS (pgtbl_ddl_test.par_gen);
+(1 row)
+
+DROP TABLE par_gen CASCADE;
+NOTICE:  drop cascades to table ch_gen
+-- User-named NOT NULL constraint must be emitted inline as
+-- "CONSTRAINT name NOT NULL" so a replay does not collide with the
+-- auto-named NOT NULL that PostgreSQL would otherwise create from a
+-- plain inline "NOT NULL".  The constraint loop skips the user-named
+-- entry to avoid double-emission.
+CREATE TABLE nn_named (
+    a int CONSTRAINT my_nn NOT NULL,
+    b int NOT NULL
+);
+SELECT * FROM pg_get_table_ddl('nn_named'::regclass, 'owner', 'false');
+                                         pg_get_table_ddl                                         
+--------------------------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.nn_named ( a integer CONSTRAINT my_nn NOT NULL, b integer NOT NULL);
+ ALTER TABLE pgtbl_ddl_test.nn_named ADD CONSTRAINT nn_named_b_not_null NOT NULL b;
+(2 rows)
+
+DROP TABLE nn_named;
+-- Locally-declared NOT NULL on an inherited column.  The column itself
+-- is omitted from the child's column list (attislocal=false), so the
+-- user-named NOT NULL must come out via the post-CREATE constraint
+-- loop rather than inline.
+CREATE TABLE nn_par (a int);
+CREATE TABLE nn_ch (b int) INHERITS (nn_par);
+ALTER TABLE nn_ch ADD CONSTRAINT my_nn NOT NULL a;
+SELECT * FROM pg_get_table_ddl('nn_ch'::regclass, 'owner', 'false');
+                                 pg_get_table_ddl                                 
+----------------------------------------------------------------------------------
+ CREATE TABLE pgtbl_ddl_test.nn_ch ( b integer) INHERITS (pgtbl_ddl_test.nn_par);
+ ALTER TABLE pgtbl_ddl_test.nn_ch ADD CONSTRAINT my_nn NOT NULL a;
+(2 rows)
+
+DROP TABLE nn_par CASCADE;
+NOTICE:  drop cascades to table nn_ch
+-- schema_qualified=false must drop the schema prefix from DEFAULT
+-- expressions and inline CHECK bodies that reference same-schema
+-- objects (sequence via nextval, function via direct call).  This
+-- relies on the deparser respecting the narrowed search_path rather
+-- than on substring stripping.
+CREATE SCHEMA pgtbl_ddl_xref;
+CREATE SEQUENCE pgtbl_ddl_xref.myseq;
+CREATE FUNCTION pgtbl_ddl_xref.f(int) RETURNS int LANGUAGE sql IMMUTABLE
+    AS 'SELECT $1';
+CREATE TABLE pgtbl_ddl_xref.t (
+    id  int DEFAULT nextval('pgtbl_ddl_xref.myseq'),
+    val int,
+    CONSTRAINT chk CHECK (pgtbl_ddl_xref.f(val) > 0)
+);
+SELECT * FROM pg_get_table_ddl('pgtbl_ddl_xref.t'::regclass,
+                               'owner', 'false',
+                               'schema_qualified', 'false');
+                                                  pg_get_table_ddl                                                  
+--------------------------------------------------------------------------------------------------------------------
+ CREATE TABLE t ( id integer DEFAULT nextval('myseq'::regclass), val integer, CONSTRAINT chk CHECK ((f(val) > 0)));
+(1 row)
+
+DROP SCHEMA pgtbl_ddl_xref CASCADE;
+NOTICE:  drop cascades to 3 other objects
+DETAIL:  drop cascades to sequence pgtbl_ddl_xref.myseq
+drop cascades to function pgtbl_ddl_xref.f(integer)
+drop cascades to table pgtbl_ddl_xref.t
+-- schema_qualified=false must NOT corrupt string literals whose
+-- contents happen to contain the schema name followed by a dot.  The
+-- in-string state of the prefix stripper preserves them verbatim.
+CREATE SCHEMA pgtbl_ddl_str;
+CREATE TABLE pgtbl_ddl_str.p (id int, note text) PARTITION BY RANGE (id);
+CREATE TABLE pgtbl_ddl_str.pc PARTITION OF pgtbl_ddl_str.p
+    (CONSTRAINT chk CHECK (note <> 'pgtbl_ddl_str.secret'))
+    FOR VALUES FROM (0) TO (100);
+SELECT * FROM pg_get_table_ddl('pgtbl_ddl_str.pc'::regclass,
+                               'owner', 'false',
+                               'schema_qualified', 'false');
+                                 pg_get_table_ddl                                  
+-----------------------------------------------------------------------------------
+ CREATE TABLE pc PARTITION OF p FOR VALUES FROM (0) TO (100);
+ ALTER TABLE pc ADD CONSTRAINT chk CHECK ((note <> 'pgtbl_ddl_str.secret'::text));
+(2 rows)
+
+DROP SCHEMA pgtbl_ddl_str CASCADE;
+NOTICE:  drop cascades to table pgtbl_ddl_str.p
+-- schema_qualified=false must NOT corrupt double-quoted identifiers
+-- whose contents happen to contain the schema name followed by a dot
+-- (for example a column literally named "<schema>.weird" inside that
+-- schema).  The in-ident state of the prefix stripper preserves them
+-- verbatim.
+CREATE SCHEMA pgtbl_ddl_qid;
+CREATE TABLE pgtbl_ddl_qid.p (id int, "pgtbl_ddl_qid.weird" int)
+    PARTITION BY RANGE (id);
+CREATE TABLE pgtbl_ddl_qid.pc PARTITION OF pgtbl_ddl_qid.p
+    (CONSTRAINT chk CHECK ("pgtbl_ddl_qid.weird" > 0))
+    FOR VALUES FROM (0) TO (100);
+SELECT * FROM pg_get_table_ddl('pgtbl_ddl_qid.pc'::regclass,
+                               'owner', 'false',
+                               'schema_qualified', 'false');
+                            pg_get_table_ddl                            
+------------------------------------------------------------------------
+ CREATE TABLE pc PARTITION OF p FOR VALUES FROM (0) TO (100);
+ ALTER TABLE pc ADD CONSTRAINT chk CHECK (("pgtbl_ddl_qid.weird" > 0));
+(2 rows)
+
+DROP SCHEMA pgtbl_ddl_qid CASCADE;
+NOTICE:  drop cascades to table pgtbl_ddl_qid.p
+-- schema_qualified=false when the base schema name itself requires
+-- quoting (its prefix starts with ").  Both forms of the prefix —
+-- bare-lowercase and quoted — must be stripped from outer
+-- ALTER TABLE / ON / CREATE INDEX prefixes, while inner quoted
+-- identifiers that happen to begin the same way must be preserved.
+CREATE SCHEMA "PgTbl-Ddl-Q";
+CREATE TABLE "PgTbl-Ddl-Q"."T" (id int, val int,
+    CONSTRAINT chk CHECK (val > 0));
+CREATE INDEX "T_val_idx" ON "PgTbl-Ddl-Q"."T" (val);
+SELECT * FROM pg_get_table_ddl('"PgTbl-Ddl-Q"."T"'::regclass,
+                               'owner', 'false',
+                               'schema_qualified', 'false');
+                                pg_get_table_ddl                                
+--------------------------------------------------------------------------------
+ CREATE TABLE "T" ( id integer, val integer, CONSTRAINT chk CHECK ((val > 0)));
+ CREATE INDEX "T_val_idx" ON "T" USING btree (val);
+(2 rows)
+
+DROP SCHEMA "PgTbl-Ddl-Q" CASCADE;
+NOTICE:  drop cascades to table "PgTbl-Ddl-Q"."T"
+-- 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..b81ff6dc4b1
--- /dev/null
+++ b/src/test/regress/sql/pg_get_table_ddl.sql
@@ -0,0 +1,440 @@
+--
+-- 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');
+
+-- schema_qualified=false strips the schema from the target table itself
+-- and from same-schema sibling references (inheritance/partition
+-- parents, identity sequences in the same schema), while cross-schema
+-- references are kept qualified for correctness.
+--
+-- Bare CREATE TABLE for an ordinary table.
+SELECT * FROM pg_get_table_ddl('basic'::regclass, 'owner', 'false',
+                               'schema_qualified', 'false');
+
+-- INHERITS parent in the same schema is emitted unqualified.
+SELECT * FROM pg_get_table_ddl('ch'::regclass, 'owner', 'false',
+                               'schema_qualified', 'false');
+
+-- Partition parent in the same schema is emitted unqualified.
+SELECT * FROM pg_get_table_ddl('parted_range_1'::regclass, 'owner', 'false',
+                               'schema_qualified', 'false');
+
+-- Cross-schema FK target keeps qualification.  The 'refd' table is in
+-- pgtbl_ddl_test; place an FK target in a different schema and verify it
+-- stays qualified even with schema_qualified=false.
+CREATE SCHEMA pgtbl_ddl_other;
+CREATE TABLE pgtbl_ddl_other.parent (id int PRIMARY KEY);
+CREATE TABLE xschema_fk (
+    id  int PRIMARY KEY,
+    pid int REFERENCES pgtbl_ddl_other.parent
+);
+SELECT * FROM pg_get_table_ddl('xschema_fk'::regclass, 'owner', 'false',
+                               'schema_qualified', 'false');
+
+-- Identity column with a custom sequence name in the same schema is
+-- emitted unqualified.
+SELECT * FROM pg_get_table_ddl('id_custom'::regclass, 'owner', 'false',
+                               'schema_qualified', 'false');
+
+-- Round-trip: with schema_qualified=false, after SET search_path the
+-- DDL can be replayed into a different target schema.
+CREATE SCHEMA pgtbl_ddl_replay;
+DO $$
+DECLARE
+    stmt text;
+BEGIN
+    SET LOCAL search_path = pgtbl_ddl_replay;
+    FOR stmt IN
+        SELECT line FROM pg_get_table_ddl('pgtbl_ddl_test.basic'::regclass,
+                                          'owner', 'false',
+                                          'schema_qualified', 'false') AS line
+    LOOP
+        EXECUTE stmt;
+    END LOOP;
+END $$;
+SELECT relnamespace::regnamespace::text, relname
+FROM pg_class WHERE oid = 'pgtbl_ddl_replay.basic'::regclass;
+DROP SCHEMA pgtbl_ddl_replay CASCADE;
+DROP TABLE xschema_fk;
+DROP SCHEMA pgtbl_ddl_other CASCADE;
+
+-- Locally-declared CHECK on a partition child has no inline column
+-- list to live in (PARTITION OF form), so it must come out as a
+-- separate ALTER TABLE ... ADD CONSTRAINT statement.
+CREATE TABLE parted_chk (id int, val int) PARTITION BY RANGE (id);
+CREATE TABLE parted_chk_child PARTITION OF parted_chk
+    (CONSTRAINT chk_inline CHECK (val > 0))
+    FOR VALUES FROM (0) TO (100);
+SELECT * FROM pg_get_table_ddl('parted_chk_child'::regclass, 'owner', 'false');
+DROP TABLE parted_chk;
+
+-- Inheritance children of a parent with a generated column should not
+-- emit an ALTER COLUMN ... SET DEFAULT for the inherited generated
+-- column: the GENERATED expression carries down through inheritance,
+-- and SET DEFAULT would fail at replay against a generated column.
+CREATE TABLE par_gen (
+    id int,
+    g  int GENERATED ALWAYS AS (id * 2) STORED
+);
+CREATE TABLE ch_gen () INHERITS (par_gen);
+SELECT * FROM pg_get_table_ddl('ch_gen'::regclass, 'owner', 'false');
+DROP TABLE par_gen CASCADE;
+
+-- User-named NOT NULL constraint must be emitted inline as
+-- "CONSTRAINT name NOT NULL" so a replay does not collide with the
+-- auto-named NOT NULL that PostgreSQL would otherwise create from a
+-- plain inline "NOT NULL".  The constraint loop skips the user-named
+-- entry to avoid double-emission.
+CREATE TABLE nn_named (
+    a int CONSTRAINT my_nn NOT NULL,
+    b int NOT NULL
+);
+SELECT * FROM pg_get_table_ddl('nn_named'::regclass, 'owner', 'false');
+DROP TABLE nn_named;
+
+-- Locally-declared NOT NULL on an inherited column.  The column itself
+-- is omitted from the child's column list (attislocal=false), so the
+-- user-named NOT NULL must come out via the post-CREATE constraint
+-- loop rather than inline.
+CREATE TABLE nn_par (a int);
+CREATE TABLE nn_ch (b int) INHERITS (nn_par);
+ALTER TABLE nn_ch ADD CONSTRAINT my_nn NOT NULL a;
+SELECT * FROM pg_get_table_ddl('nn_ch'::regclass, 'owner', 'false');
+DROP TABLE nn_par CASCADE;
+
+-- schema_qualified=false must drop the schema prefix from DEFAULT
+-- expressions and inline CHECK bodies that reference same-schema
+-- objects (sequence via nextval, function via direct call).  This
+-- relies on the deparser respecting the narrowed search_path rather
+-- than on substring stripping.
+CREATE SCHEMA pgtbl_ddl_xref;
+CREATE SEQUENCE pgtbl_ddl_xref.myseq;
+CREATE FUNCTION pgtbl_ddl_xref.f(int) RETURNS int LANGUAGE sql IMMUTABLE
+    AS 'SELECT $1';
+CREATE TABLE pgtbl_ddl_xref.t (
+    id  int DEFAULT nextval('pgtbl_ddl_xref.myseq'),
+    val int,
+    CONSTRAINT chk CHECK (pgtbl_ddl_xref.f(val) > 0)
+);
+SELECT * FROM pg_get_table_ddl('pgtbl_ddl_xref.t'::regclass,
+                               'owner', 'false',
+                               'schema_qualified', 'false');
+DROP SCHEMA pgtbl_ddl_xref CASCADE;
+
+-- schema_qualified=false must NOT corrupt string literals whose
+-- contents happen to contain the schema name followed by a dot.  The
+-- in-string state of the prefix stripper preserves them verbatim.
+CREATE SCHEMA pgtbl_ddl_str;
+CREATE TABLE pgtbl_ddl_str.p (id int, note text) PARTITION BY RANGE (id);
+CREATE TABLE pgtbl_ddl_str.pc PARTITION OF pgtbl_ddl_str.p
+    (CONSTRAINT chk CHECK (note <> 'pgtbl_ddl_str.secret'))
+    FOR VALUES FROM (0) TO (100);
+SELECT * FROM pg_get_table_ddl('pgtbl_ddl_str.pc'::regclass,
+                               'owner', 'false',
+                               'schema_qualified', 'false');
+DROP SCHEMA pgtbl_ddl_str CASCADE;
+
+-- schema_qualified=false must NOT corrupt double-quoted identifiers
+-- whose contents happen to contain the schema name followed by a dot
+-- (for example a column literally named "<schema>.weird" inside that
+-- schema).  The in-ident state of the prefix stripper preserves them
+-- verbatim.
+CREATE SCHEMA pgtbl_ddl_qid;
+CREATE TABLE pgtbl_ddl_qid.p (id int, "pgtbl_ddl_qid.weird" int)
+    PARTITION BY RANGE (id);
+CREATE TABLE pgtbl_ddl_qid.pc PARTITION OF pgtbl_ddl_qid.p
+    (CONSTRAINT chk CHECK ("pgtbl_ddl_qid.weird" > 0))
+    FOR VALUES FROM (0) TO (100);
+SELECT * FROM pg_get_table_ddl('pgtbl_ddl_qid.pc'::regclass,
+                               'owner', 'false',
+                               'schema_qualified', 'false');
+DROP SCHEMA pgtbl_ddl_qid CASCADE;
+
+-- schema_qualified=false when the base schema name itself requires
+-- quoting (its prefix starts with ").  Both forms of the prefix —
+-- bare-lowercase and quoted — must be stripped from outer
+-- ALTER TABLE / ON / CREATE INDEX prefixes, while inner quoted
+-- identifiers that happen to begin the same way must be preserved.
+CREATE SCHEMA "PgTbl-Ddl-Q";
+CREATE TABLE "PgTbl-Ddl-Q"."T" (id int, val int,
+    CONSTRAINT chk CHECK (val > 0));
+CREATE INDEX "T_val_idx" ON "PgTbl-Ddl-Q"."T" (val);
+SELECT * FROM pg_get_table_ddl('"PgTbl-Ddl-Q"."T"'::regclass,
+                               'owner', 'false',
+                               'schema_qualified', 'false');
+DROP SCHEMA "PgTbl-Ddl-Q" CASCADE;
+
+-- 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

