diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 082a3575d6..29990ddc96 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -19272,6 +19272,25 @@ remove_on_commit_action(Oid relid)
 	}
 }
 
+/*
+ * Return the ON COMMIT action set for a relation, or NOOP if none
+ * is registered
+*/
+OnCommitAction
+get_on_commit_action(Oid relid)
+{
+	ListCell   *l;
+
+	foreach(l, on_commits)
+	{
+		OnCommitItem *oc = (OnCommitItem *) lfirst(l);
+
+		if (oc->relid == relid)
+			return oc->oncommit;
+	}
+	return ONCOMMIT_NOOP;
+}
+
 /*
  * Perform ON COMMIT actions.
  *
diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c
index 3d6e6bdbfd..13f2d93371 100644
--- a/src/backend/utils/adt/ruleutils.c
+++ b/src/backend/utils/adt/ruleutils.c
@@ -20,15 +20,20 @@
 #include <fcntl.h>
 
 #include "access/amapi.h"
+#include "access/heaptoast.h"
 #include "access/htup_details.h"
 #include "access/relation.h"
 #include "access/table.h"
+#include "access/toast_compression.h"
+#include "catalog/partition.h"
 #include "catalog/pg_aggregate.h"
 #include "catalog/pg_am.h"
+#include "catalog/pg_attrdef.h"
 #include "catalog/pg_authid.h"
 #include "catalog/pg_collation.h"
 #include "catalog/pg_constraint.h"
 #include "catalog/pg_depend.h"
+#include "catalog/pg_inherits.h"
 #include "catalog/pg_language.h"
 #include "catalog/pg_opclass.h"
 #include "catalog/pg_operator.h"
@@ -38,6 +43,7 @@
 #include "catalog/pg_trigger.h"
 #include "catalog/pg_type.h"
 #include "commands/defrem.h"
+#include "commands/tablecmds.h"
 #include "commands/tablespace.h"
 #include "common/keywords.h"
 #include "executor/spi.h"
@@ -13707,3 +13713,683 @@ get_range_partbound_string(List *bound_datums)
 
 	return buf->data;
 }
+
+/*
+ * pg_get_table_ddl
+ *
+ * Returns the CREATE TABLE DDL statement for the specified table.
+ * Handles regular, temporary, unlogged, and partitioned tables.
+ */
+Datum
+pg_get_table_ddl(PG_FUNCTION_ARGS)
+{
+	Oid			relid = PG_GETARG_OID(0);
+	StringInfoData buf;
+	Relation	rel;
+	TupleDesc	tupdesc;
+	Form_pg_class relform;
+	char	   *tablename;
+	char	   *schemaname;
+	char	   *tablespacename = NULL;
+	bool		first_col = true;
+	int			i;
+
+	/* Open the relation */
+	rel = relation_open(relid, AccessShareLock);
+	relform = RelationGetForm(rel);
+
+	/* Only handle tables and partitioned tables */
+	if (relform->relkind != RELKIND_RELATION &&
+		relform->relkind != RELKIND_PARTITIONED_TABLE)
+	{
+		relation_close(rel, AccessShareLock);
+		ereport(ERROR,
+				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+				 errmsg("relation \"%s\" is not a table",
+						RelationGetRelationName(rel))));
+	}
+
+	/* Get tuple descriptor */
+	tupdesc = RelationGetDescr(rel);
+
+	initStringInfo(&buf);
+
+	/* Get table and schema names */
+	tablename = RelationGetRelationName(rel);
+	schemaname = get_namespace_name(RelationGetNamespace(rel));
+
+	/* Start building the CREATE TABLE statement */
+	appendStringInfo(&buf, "CREATE ");
+
+	/* Add table type modifiers */
+	if (relform->relpersistence == RELPERSISTENCE_UNLOGGED)
+		appendStringInfoString(&buf, "UNLOGGED ");
+	else if (relform->relpersistence == RELPERSISTENCE_TEMP)
+		appendStringInfoString(&buf, "TEMPORARY ");
+
+	appendStringInfoString(&buf, "TABLE ");
+
+	/* Add schema-qualified table name */
+	if (relform->relpersistence == RELPERSISTENCE_TEMP)
+		appendStringInfo(&buf, "%s",
+						 quote_identifier(tablename));
+	else
+		appendStringInfo(&buf, "%s.%s",
+						 quote_identifier(schemaname),
+						 quote_identifier(tablename));
+
+	/* Handle typed tables */
+	if (OidIsValid(relform->reloftype))
+	{
+		/*
+		 * For typed tables, we don't need full column definitions, just any
+		 * local columns or constraints
+		 */
+		bool		has_local_columns = false;
+		char	   *typename = format_type_be(relform->reloftype);
+
+		appendStringInfo(&buf, " OF %s", typename);
+
+		/* Check if there are any local columns beyond the type */
+		for (i = 0; i < tupdesc->natts; i++)
+		{
+			Form_pg_attribute attr = TupleDescAttr(tupdesc, i);
+
+			if (attr->attisdropped || attr->attinhcount > 0)
+				continue;
+			has_local_columns = true;
+			break;
+		}
+
+		if (has_local_columns || (rel->rd_att->constr && rel->rd_att->constr->num_check > 0))
+		{
+			appendStringInfoString(&buf, "\n(\n");
+			first_col = true;
+
+			/* Add any local columns */
+			for (i = 0; i < tupdesc->natts; i++)
+			{
+				Form_pg_attribute attr = TupleDescAttr(tupdesc, i);
+
+				if (attr->attisdropped || attr->attinhcount > 0)
+					continue;
+
+				if (!first_col)
+					appendStringInfoString(&buf, ",\n");
+				first_col = false;
+
+				/* Only show local modifications/constraints for this column */
+				appendStringInfo(&buf, "    %s", quote_identifier(NameStr(attr->attname)));
+
+				if (attr->attnotnull)
+					appendStringInfoString(&buf, " NOT NULL");
+
+				if (attr->atthasdef)
+				{
+					AttrDefault *defval = NULL;
+
+					if (rel->rd_att->constr && rel->rd_att->constr->defval)
+					{
+						for (int j = 0; j < rel->rd_att->constr->num_defval; j++)
+						{
+							if (rel->rd_att->constr->defval[j].adnum == attr->attnum)
+							{
+								defval = &rel->rd_att->constr->defval[j];
+								break;
+							}
+						}
+					}
+
+					if (defval)
+						appendStringInfo(&buf, " DEFAULT %s",
+										 text_to_cstring(pg_get_expr_worker(cstring_to_text(defval->adbin), relid, 0)));
+				}
+			}
+
+			/* Add CHECK constraints for typed tables */
+			if (rel->rd_att->constr && rel->rd_att->constr->num_check > 0)
+			{
+				for (i = 0; i < rel->rd_att->constr->num_check; i++)
+				{
+					ConstrCheck *check = &rel->rd_att->constr->check[i];
+
+					if (!first_col)
+						appendStringInfoString(&buf, ",\n");
+					first_col = false;
+					appendStringInfo(&buf, "    CONSTRAINT %s CHECK (%s)",
+									 quote_identifier(check->ccname),
+									 text_to_cstring(pg_get_expr_worker(cstring_to_text(check->ccbin), relid, 0)));
+				}
+			}
+
+			appendStringInfoString(&buf, "\n)");
+		}
+	}
+	/* Handle partition tables - only show local columns if any */
+	else if (rel->rd_rel->relispartition)
+	{
+		/*
+		 * For partitions, we typically don't show inherited columns, only
+		 * local additions or modifications
+		 */
+		bool		has_local_additions = false;
+
+		/* Check for any local columns or constraints */
+		for (i = 0; i < tupdesc->natts; i++)
+		{
+			Form_pg_attribute attr = TupleDescAttr(tupdesc, i);
+
+			if (attr->attisdropped)
+				continue;
+			/* Check if this is a local column (not inherited) */
+			if (attr->attinhcount == 0)
+			{
+				has_local_additions = true;
+				break;
+			}
+		}
+
+		if (has_local_additions || (rel->rd_att->constr && rel->rd_att->constr->num_check > 0))
+		{
+			appendStringInfoString(&buf, "\n(\n");
+			first_col = true;
+
+			/* Add local columns only */
+			for (i = 0; i < tupdesc->natts; i++)
+			{
+				Form_pg_attribute attr = TupleDescAttr(tupdesc, i);
+				char	   *typename_with_typemod;
+
+				if (attr->attisdropped || attr->attinhcount > 0)
+					continue;
+
+				if (!first_col)
+					appendStringInfoString(&buf, ",\n");
+				first_col = false;
+
+				/* Full column definition for local columns */
+				appendStringInfo(&buf, "    %s", quote_identifier(NameStr(attr->attname)));
+
+				typename_with_typemod = format_type_with_typemod(attr->atttypid, attr->atttypmod);
+				appendStringInfo(&buf, " %s", typename_with_typemod);
+
+				if (attr->attnotnull)
+					appendStringInfoString(&buf, " NOT NULL");
+
+				if (attr->atthasdef)
+				{
+					AttrDefault *defval = NULL;
+
+					if (rel->rd_att->constr && rel->rd_att->constr->defval)
+					{
+						for (int j = 0; j < rel->rd_att->constr->num_defval; j++)
+						{
+							if (rel->rd_att->constr->defval[j].adnum == attr->attnum)
+							{
+								defval = &rel->rd_att->constr->defval[j];
+								break;
+							}
+						}
+					}
+
+					if (defval)
+						appendStringInfo(&buf, " DEFAULT %s",
+										 text_to_cstring(pg_get_expr_worker(cstring_to_text(defval->adbin), relid, 0)));
+				}
+			}
+
+			/* Add local CHECK constraints */
+			if (rel->rd_att->constr && rel->rd_att->constr->num_check > 0)
+			{
+				for (i = 0; i < rel->rd_att->constr->num_check; i++)
+				{
+					ConstrCheck *check = &rel->rd_att->constr->check[i];
+
+					if (!first_col)
+						appendStringInfoString(&buf, ",\n");
+					first_col = false;
+					appendStringInfo(&buf, "    CONSTRAINT %s CHECK (%s)",
+									 quote_identifier(check->ccname),
+									 text_to_cstring(pg_get_expr_worker(cstring_to_text(check->ccbin), relid, 0)));
+				}
+			}
+
+			appendStringInfoString(&buf, "\n)");
+		}
+	}
+	/* Handle regular tables and partitioned tables */
+	else
+	{
+		appendStringInfoString(&buf, "\n(\n");
+
+		/* Process each column */
+		for (i = 0; i < tupdesc->natts; i++)
+		{
+			Form_pg_attribute attr = TupleDescAttr(tupdesc, i);
+			char	   *typename_with_typemod;
+			bool		notnull;
+			char		storage_type;
+
+			/* Skip dropped columns */
+			if (attr->attisdropped)
+				continue;
+
+			if (!first_col)
+				appendStringInfoString(&buf, ",\n");
+			first_col = false;
+
+			/* Column name */
+			appendStringInfo(&buf, "    %s", quote_identifier(NameStr(attr->attname)));
+
+			/* Data type with type modifiers */
+			typename_with_typemod = format_type_with_typemod(attr->atttypid, attr->atttypmod);
+			appendStringInfo(&buf, " %s", typename_with_typemod);
+
+			/* Collation */
+			if (OidIsValid(attr->attcollation))
+			{
+				HeapTuple	coll_tuple;
+				Form_pg_collation coll_form;
+
+				coll_tuple = SearchSysCache1(COLLOID, ObjectIdGetDatum(attr->attcollation));
+				if (HeapTupleIsValid(coll_tuple))
+				{
+					/*
+					 * Only show collation if it's not the default for the
+					 * type
+					 */
+					Oid			default_collation = get_typcollation(attr->atttypid);
+
+					coll_form = (Form_pg_collation) GETSTRUCT(coll_tuple);
+
+					if (attr->attcollation != default_collation)
+					{
+						char	   *coll_namespace = get_namespace_name(coll_form->collnamespace);
+
+						if (coll_namespace && strcmp(coll_namespace, "pg_catalog") != 0)
+							appendStringInfo(&buf, " COLLATE %s.%s",
+											 quote_identifier(coll_namespace),
+											 quote_identifier(NameStr(coll_form->collname)));
+						else
+							appendStringInfo(&buf, " COLLATE %s",
+											 quote_identifier(NameStr(coll_form->collname)));
+					}
+					ReleaseSysCache(coll_tuple);
+				}
+			}
+
+			/* NOT NULL constraint */
+			notnull = attr->attnotnull;
+			if (notnull)
+				appendStringInfoString(&buf, " NOT NULL");
+
+			/* Default expression */
+			if (attr->atthasdef)
+			{
+				AttrDefault *defval = NULL;
+
+				if (rel->rd_att->constr && rel->rd_att->constr->defval)
+				{
+					for (int j = 0; j < rel->rd_att->constr->num_defval; j++)
+					{
+						if (rel->rd_att->constr->defval[j].adnum == attr->attnum)
+						{
+							defval = &rel->rd_att->constr->defval[j];
+							break;
+						}
+					}
+				}
+
+				if (defval)
+				{
+					appendStringInfo(&buf, " DEFAULT %s",
+									 text_to_cstring(pg_get_expr_worker(cstring_to_text(defval->adbin), relid, 0)));
+				}
+			}
+
+			/* Storage type */
+			storage_type = attr->attstorage;
+			if (storage_type != get_typstorage(attr->atttypid))
+			{
+				switch (storage_type)
+				{
+					case 'p':
+						appendStringInfoString(&buf, " STORAGE PLAIN");
+						break;
+					case 'e':
+						appendStringInfoString(&buf, " STORAGE EXTERNAL");
+						break;
+					case 'm':
+						appendStringInfoString(&buf, " STORAGE MAIN");
+						break;
+					case 'x':
+						appendStringInfoString(&buf, " STORAGE EXTENDED");
+						break;
+				}
+			}
+
+			/* Compression method (PostgreSQL 14+) */
+			if (attr->attcompression != InvalidCompressionMethod)
+			{
+				const char *compression_name = GetCompressionMethodName(attr->attcompression);
+
+				if (compression_name)
+					appendStringInfo(&buf, " COMPRESSION %s", compression_name);
+			}
+		}
+
+		/* Add table constraints (PRIMARY KEY, UNIQUE, CHECK, etc.) */
+		if (rel->rd_att->constr && rel->rd_att->constr->num_check > 0)
+		{
+			for (i = 0; i < rel->rd_att->constr->num_check; i++)
+			{
+				ConstrCheck *check = &rel->rd_att->constr->check[i];
+
+				appendStringInfo(&buf, ",\n    CONSTRAINT %s CHECK (%s)",
+								 quote_identifier(check->ccname),
+								 text_to_cstring(pg_get_expr_worker(cstring_to_text(check->ccbin), relid, 0)));
+			}
+		}
+
+		/* Add indexes as constraints (PRIMARY KEY, UNIQUE) */
+		{
+			List	   *indexoidlist = RelationGetIndexList(rel);
+			ListCell   *l;
+
+			foreach(l, indexoidlist)
+			{
+				Oid			indexoid = lfirst_oid(l);
+				Relation	indexrel;
+				Form_pg_index indexform;
+				HeapTuple	indexTuple;
+
+				indexTuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(indexoid));
+				if (!HeapTupleIsValid(indexTuple))
+					continue;
+
+				indexform = (Form_pg_index) GETSTRUCT(indexTuple);
+
+				if (indexform->indisprimary || indexform->indisunique)
+				{
+					indexrel = relation_open(indexoid, AccessShareLock);
+
+					if (indexform->indisprimary)
+						appendStringInfoString(&buf, ",\n    PRIMARY KEY (");
+					else
+						appendStringInfoString(&buf, ",\n    UNIQUE (");
+
+					/* Add index column names */
+					for (int j = 0; j < indexform->indnatts; j++)
+					{
+						int16		attnum = indexform->indkey.values[j];
+						char	   *attname = get_attname(relid, attnum, false);
+
+						if (j > 0)
+							appendStringInfoString(&buf, ", ");
+						appendStringInfo(&buf, "%s", quote_identifier(attname));
+					}
+					appendStringInfoString(&buf, ")");
+
+					relation_close(indexrel, AccessShareLock);
+				}
+
+				ReleaseSysCache(indexTuple);
+			}
+
+			list_free(indexoidlist);
+		}
+
+		appendStringInfoString(&buf, "\n)");
+	}
+
+	/* Add INHERITS clause for inheritance (not partitioning) */
+	if (rel->rd_rel->relhassubclass == false && rel->rd_rel->relispartition == false)
+	{
+		Relation	catalogrel;
+		SysScanDesc scan;
+		ScanKeyData key;
+		HeapTuple	inheritsTuple;
+		bool		first_parent = true;
+
+		catalogrel = table_open(InheritsRelationId, AccessShareLock);
+
+		ScanKeyInit(&key,
+					Anum_pg_inherits_inhrelid,
+					BTEqualStrategyNumber, F_OIDEQ,
+					ObjectIdGetDatum(relid));
+
+		scan = systable_beginscan(catalogrel, InheritsRelidSeqnoIndexId,
+								  true, NULL, 1, &key);
+
+		while (HeapTupleIsValid(inheritsTuple = systable_getnext(scan)))
+		{
+			Form_pg_inherits inh = (Form_pg_inherits) GETSTRUCT(inheritsTuple);
+			char	   *parent_name;
+			char	   *parent_schema;
+
+			if (first_parent)
+			{
+				appendStringInfoString(&buf, " INHERITS (");
+				first_parent = false;
+			}
+			else
+				appendStringInfoString(&buf, ", ");
+
+			parent_name = get_rel_name(inh->inhparent);
+			parent_schema = get_namespace_name(get_rel_namespace(inh->inhparent));
+
+			if (parent_schema && strcmp(parent_schema, "public") != 0)
+				appendStringInfo(&buf, "%s.%s",
+								 quote_identifier(parent_schema),
+								 quote_identifier(parent_name));
+			else
+				appendStringInfo(&buf, "%s", quote_identifier(parent_name));
+		}
+
+		if (!first_parent)
+			appendStringInfoString(&buf, ")");
+
+		systable_endscan(scan);
+		table_close(catalogrel, AccessShareLock);
+	}
+
+	/* Add PARTITION OF clause for partitioned tables */
+	if (rel->rd_rel->relispartition)
+	{
+		char	   *parent_name;
+		char	   *parent_schema;
+		Oid			parent_oid = get_partition_parent(relid, false);
+		HeapTuple	reltuple;
+		text	   *boundDatum;
+
+
+		/* Get pg_class.relpartbound */
+		reltuple = SearchSysCache1(RELOID,
+								   ObjectIdGetDatum(RelationGetRelid(rel)));
+		if (!HeapTupleIsValid(reltuple))
+			elog(ERROR, "cache lookup failed for relation %u", relid);
+
+
+		/* There must be a partition bound (XXX I think) */
+		boundDatum = DatumGetTextPP(SysCacheGetAttrNotNull(RELOID, reltuple,
+														   Anum_pg_class_relpartbound));
+		if (OidIsValid(parent_oid))
+		{
+			parent_name = get_rel_name(parent_oid);
+			parent_schema = get_namespace_name(get_rel_namespace(parent_oid));
+
+			appendStringInfoString(&buf, " PARTITION OF ");
+			appendStringInfo(&buf, "%s.%s",
+							 quote_identifier(parent_schema),
+							 quote_identifier(parent_name));
+
+			appendStringInfo(&buf, " %s", text_to_cstring(pg_get_expr_worker(boundDatum, relid, 0)));
+		}
+
+		ReleaseSysCache(reltuple);
+	}
+
+	/* Add PARTITION BY clause for partitioned tables */
+
+	/* XXX check if this handles all cases */
+
+	if (relform->relkind == RELKIND_PARTITIONED_TABLE)
+	{
+		PartitionKey partkey = RelationGetPartitionKey(rel);
+		char	   *parttype;
+
+		switch (partkey->strategy)
+		{
+			case PARTITION_STRATEGY_RANGE:
+				parttype = "RANGE";
+				break;
+			case PARTITION_STRATEGY_LIST:
+				parttype = "LIST";
+				break;
+			case PARTITION_STRATEGY_HASH:
+				parttype = "HASH";
+				break;
+			default:
+				parttype = "UNKNOWN";
+				break;
+		}
+
+		appendStringInfo(&buf, " PARTITION BY %s (", parttype);
+
+		/* Add partition key columns */
+		for (i = 0; i < partkey->partnatts; i++)
+		{
+			AttrNumber	attnum = partkey->partattrs[i];
+			char	   *attname;
+
+			if (i > 0)
+				appendStringInfoString(&buf, ", ");
+
+			if (attnum > 0)
+			{
+				/* Regular column */
+				attname = get_attname(relid, attnum, false);
+				appendStringInfo(&buf, "%s", quote_identifier(attname));
+			}
+			else
+			{
+				/* Expression */
+				Node	   *partexpr = list_nth(partkey->partexprs,
+												partkey->partattrs[i] - 1 - FirstLowInvalidHeapAttributeNumber);
+				char	   *exprstr = deparse_expression(partexpr,
+														 deparse_context_for(RelationGetRelationName(rel), relid),
+														 false, false);
+
+				appendStringInfo(&buf, "(%s)", exprstr);
+			}
+
+			/* Add collation if specified */
+			if (OidIsValid(partkey->partcollation[i]))
+			{
+				char	   *collname = get_collation_name(partkey->partcollation[i]);
+
+				if (collname && strcmp(collname, "default") != 0)
+					appendStringInfo(&buf, " COLLATE %s", quote_identifier(collname));
+			}
+		}
+
+		appendStringInfoString(&buf, ")");
+	}
+
+	/* Add WITH options */
+	{
+		bool		first_option = true;
+
+		/* Fill factor - use RelationGetFillFactor */
+		if (relform->relkind == RELKIND_RELATION)
+		{
+			int			fillfactor = RelationGetFillFactor(rel, HEAP_DEFAULT_FILLFACTOR);
+
+			if (fillfactor != HEAP_DEFAULT_FILLFACTOR)
+			{
+				if (first_option)
+				{
+					appendStringInfoString(&buf, " WITH (");
+					first_option = false;
+				}
+				else
+					appendStringInfoString(&buf, ", ");
+
+				appendStringInfo(&buf, "fillfactor=%d", fillfactor);
+			}
+		}
+
+		/* Toast tuple target */
+		if (relform->reltoastrelid != InvalidOid)
+		{
+			Relation	toast_rel = relation_open(relform->reltoastrelid, AccessShareLock);
+
+			if (toast_rel->rd_options)
+			{
+				StdRdOptions *toast_options = (StdRdOptions *) toast_rel->rd_options;
+
+				/* XXX check if this is the right test */
+				if (toast_options->toast_tuple_target != TOAST_TUPLE_TARGET)
+				{
+					if (first_option)
+					{
+						appendStringInfoString(&buf, " WITH (");
+						first_option = false;
+					}
+					else
+						appendStringInfoString(&buf, ", ");
+
+					appendStringInfo(&buf, "toast_tuple_target=%d",
+									 toast_options->toast_tuple_target);
+				}
+			}
+			relation_close(toast_rel, AccessShareLock);
+		}
+
+		if (!first_option)
+			appendStringInfoString(&buf, ")");
+	}
+
+	/* Add tablespace */
+	if (OidIsValid(relform->reltablespace))
+	{
+		tablespacename = get_tablespace_name(relform->reltablespace);
+		if (tablespacename)
+			appendStringInfo(&buf, " TABLESPACE %s", quote_identifier(tablespacename));
+	}
+
+	/* Add ON COMMIT clause for temporary tables */
+	if (relform->relpersistence == RELPERSISTENCE_TEMP)
+	{
+		/* Get ON COMMIT from local oncommits structure */
+		OnCommitAction oncommit = get_on_commit_action(relid);
+
+		/*
+		 * Only add ON COMMIT clause if we successfully determined the
+		 * behavior
+		 */
+		switch (oncommit)
+		{
+			case ONCOMMIT_DELETE_ROWS:
+				appendStringInfoString(&buf, " ON COMMIT DELETE ROWS");
+				break;
+			case ONCOMMIT_PRESERVE_ROWS:
+				appendStringInfoString(&buf, " ON COMMIT PRESERVE ROWS");
+				break;
+			case ONCOMMIT_DROP:
+				appendStringInfoString(&buf, " ON COMMIT DROP");
+				break;
+			case ONCOMMIT_NOOP:
+			default:
+				/* Don't add anything for default behavior */
+				break;
+		}
+	}
+
+	appendStringInfoString(&buf, ";");
+
+	/* Close the relation */
+	relation_close(rel, AccessShareLock);
+
+	PG_RETURN_TEXT_P(cstring_to_text(buf.data));
+}
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 118d6da1ac..8c4cf24b62 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -12575,5 +12575,7 @@
   proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
   proargnames => '{pid,io_id,io_generation,state,operation,off,length,target,handle_data_len,raw_result,result,target_desc,f_sync,f_localmem,f_buffered}',
   prosrc => 'pg_get_aios' },
-
+{ oid => '8888', descr => 'show CREATE TABLE command for table',
+  proname => 'pg_get_table_ddl', prorettype => 'text',
+  proargtypes => 'regclass', prosrc => 'pg_get_table_ddl' },
 ]
diff --git a/src/include/commands/tablecmds.h b/src/include/commands/tablecmds.h
index 6832470d38..aa5ff4a93f 100644
--- a/src/include/commands/tablecmds.h
+++ b/src/include/commands/tablecmds.h
@@ -91,6 +91,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/object_ddl.out b/src/test/regress/expected/object_ddl.out
new file mode 100644
index 0000000000..89a5858675
--- /dev/null
+++ b/src/test/regress/expected/object_ddl.out
@@ -0,0 +1,45 @@
+create table foo (t text not null, i serial primary key, check (t > 'a'));
+create temp table bar (t text not null) on commit delete rows;
+\pset format unaligned
+select pg_get_table_ddl('foo');
+pg_get_table_ddl
+CREATE TABLE public.foo
+(
+    t text NOT NULL,
+    i integer NOT NULL DEFAULT nextval('foo_i_seq'::regclass),
+    CONSTRAINT foo_t_check CHECK ((t > 'a'::text)),
+    PRIMARY KEY (i)
+);
+(1 row)
+select pg_get_table_ddl('bar');
+pg_get_table_ddl
+CREATE TEMPORARY TABLE bar
+(
+    t text NOT NULL
+) ON COMMIT DELETE ROWS;
+(1 row)
+CREATE TABLE test_part_parent (
+    a integer DEFAULT 2501,
+    b integer DEFAULT 142857
+)
+PARTITION BY LIST (a);
+create table test_part_partition_123 partition of test_part_parent for values in (1,2,3);
+create table test_part_partition_def partition of test_part_parent default;
+select pg_get_table_ddl('test_part_parent');
+pg_get_table_ddl
+CREATE TABLE public.test_part_parent
+(
+    a integer DEFAULT 2501,
+    b integer DEFAULT 142857
+) PARTITION BY LIST (a);
+(1 row)
+select pg_get_table_ddl('test_part_partition_123');
+pg_get_table_ddl
+CREATE TABLE public.test_part_partition_123 PARTITION OF public.test_part_parent FOR VALUES IN (1, 2, 3);
+(1 row)
+select pg_get_table_ddl('test_part_partition_def');
+pg_get_table_ddl
+CREATE TABLE public.test_part_partition_def PARTITION OF public.test_part_parent DEFAULT;
+(1 row)
+drop table foo;
+drop table bar;
diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule
index fbffc67ae6..1e5630eaa5 100644
--- a/src/test/regress/parallel_schedule
+++ b/src/test/regress/parallel_schedule
@@ -43,7 +43,7 @@ test: copy copyselect copydml copyencoding insert insert_conflict
 # Note: many of the tests in later groups depend on create_index
 # ----------
 test: create_function_c create_misc create_operator create_procedure create_table create_type create_schema
-test: create_index create_index_spgist create_view index_including index_including_gist
+test: create_index create_index_spgist create_view index_including index_including_gist object_ddl
 
 # ----------
 # Another group of parallel tests
diff --git a/src/test/regress/sql/object_ddl.sql b/src/test/regress/sql/object_ddl.sql
new file mode 100644
index 0000000000..f09c24b800
--- /dev/null
+++ b/src/test/regress/sql/object_ddl.sql
@@ -0,0 +1,31 @@
+create table foo (t text not null, i serial primary key, check (t > 'a'));
+
+create temp table bar (t text not null) on commit delete rows;
+
+
+\pset format unaligned
+
+select pg_get_table_ddl('foo');
+
+select pg_get_table_ddl('bar');
+
+CREATE TABLE test_part_parent (
+    a integer DEFAULT 2501,
+    b integer DEFAULT 142857
+)
+PARTITION BY LIST (a);
+
+create table test_part_partition_123 partition of test_part_parent for values in (1,2,3);
+
+create table test_part_partition_def partition of test_part_parent default;
+
+select pg_get_table_ddl('test_part_parent');
+
+select pg_get_table_ddl('test_part_partition_123');
+
+select pg_get_table_ddl('test_part_partition_def');
+
+
+drop table foo;
+
+drop table bar;