[#24078] xCluster: support user-defined composite types in automatic mode

Summary:
This extends the automatic mode OID preserving machinery to also preserve the following OIDs:
  * pg_type OID for shell types
  * pg_type OID for single-range types
  * pg_type OID for user-defined composite types
  * pg_type OID for user-defined base types

This allows user-defined composite types, single-range types, and user-defined based types to be created in automatic mode while replication is running.

This diff does not preserve OIDs for multi-range types but we do not currently support storing those in tables anyways.

Note that table row types are also considered composite types; this diff does not preserve their OIDs.  Also not preserved: pg_type OIDs for array types and domain types.

Because enum pg_type OID's are now preserved I was able to remove the kludge where we ignored the enum pg_type OID when locating enum labels.

Did some refactoring to allow combining the new type OID map with the existing sequence OID map since they have the same fields.

Fixes #24078
Jira: DB-12971

Test Plan:
Added a new regression test for preserving pg_type OIDs for user-defined composite types, non-array-based types, enums, ranges, and shell types:
```
ybd release --cxx-test xcluster_ddl_replication_pgregress-test --gtest_filter '*.PgRegressCreateDropType'
```

The same preservation test is applied to the other regression tests:
```
ybd release --cxx-test xcluster_ddl_replication_pgregress-test
```

For reviewer convenience, the following is from the logs running the new
test:
```
I0327 12:35:45.720230  4215 xcluster_ddl_replication_pgregress-test.cc:148] pg_type OIDs that must match on both sides are:
pg_catalog, aclitem, 1033, b, NULL
pg_catalog, bit, 1560, b, NULL
pg_catalog, bool, 16, b, NULL
pg_catalog, box, 603, b, NULL
pg_catalog, bpchar, 1042, b, NULL
pg_catalog, bytea, 17, b, NULL
pg_catalog, char, 18, b, NULL
pg_catalog, cid, 29, b, NULL
pg_catalog, cidr, 650, b, NULL
pg_catalog, circle, 718, b, NULL
pg_catalog, date, 1082, b, NULL
pg_catalog, daterange, 3912, r, NULL
pg_catalog, float4, 700, b, NULL
pg_catalog, float8, 701, b, NULL
pg_catalog, gtsvector, 3642, b, NULL
pg_catalog, inet, 869, b, NULL
pg_catalog, int2, 21, b, NULL
pg_catalog, int4, 23, b, NULL
pg_catalog, int4range, 3904, r, NULL
pg_catalog, int8, 20, b, NULL
pg_catalog, int8range, 3926, r, NULL
pg_catalog, interval, 1186, b, NULL
pg_catalog, json, 114, b, NULL
pg_catalog, jsonb, 3802, b, NULL
pg_catalog, jsonpath, 8003, b, NULL
pg_catalog, line, 628, b, NULL
pg_catalog, lseg, 601, b, NULL
pg_catalog, macaddr, 829, b, NULL
pg_catalog, macaddr8, 774, b, NULL
pg_catalog, money, 790, b, NULL
pg_catalog, name, 19, b, NULL
pg_catalog, numeric, 1700, b, NULL
pg_catalog, numrange, 3906, r, NULL
pg_catalog, oid, 26, b, NULL
pg_catalog, path, 602, b, NULL
pg_catalog, pg_brin_bloom_summary, 4600, b, NULL
pg_catalog, pg_brin_minmax_multi_summary, 4601, b, NULL
pg_catalog, pg_dependencies, 3402, b, NULL
pg_catalog, pg_lsn, 3220, b, NULL
pg_catalog, pg_mcv_list, 5017, b, NULL
pg_catalog, pg_ndistinct, 3361, b, NULL
pg_catalog, pg_node_tree, 194, b, NULL
pg_catalog, pg_snapshot, 5038, b, NULL
pg_catalog, point, 600, b, NULL
pg_catalog, polygon, 604, b, NULL
pg_catalog, refcursor, 1790, b, NULL
pg_catalog, regclass, 2205, b, NULL
pg_catalog, regcollation, 4191, b, NULL
pg_catalog, regconfig, 3734, b, NULL
pg_catalog, regdictionary, 3769, b, NULL
pg_catalog, regnamespace, 4089, b, NULL
pg_catalog, regoper, 2203, b, NULL
pg_catalog, regoperator, 2204, b, NULL
pg_catalog, regproc, 24, b, NULL
pg_catalog, regprocedure, 2202, b, NULL
pg_catalog, regrole, 4096, b, NULL
pg_catalog, regtype, 2206, b, NULL
pg_catalog, text, 25, b, NULL
pg_catalog, tid, 27, b, NULL
pg_catalog, time, 1083, b, NULL
pg_catalog, timestamp, 1114, b, NULL
pg_catalog, timestamptz, 1184, b, NULL
pg_catalog, timetz, 1266, b, NULL
pg_catalog, tsquery, 3615, b, NULL
pg_catalog, tsrange, 3908, r, NULL
pg_catalog, tstzrange, 3910, r, NULL
pg_catalog, tsvector, 3614, b, NULL
pg_catalog, txid_snapshot, 2970, b, NULL
pg_catalog, uuid, 2950, b, NULL
pg_catalog, varbit, 1562, b, NULL
pg_catalog, varchar, 1043, b, NULL
pg_catalog, xid, 28, b, NULL
pg_catalog, xid8, 5069, b, NULL
pg_catalog, xml, 142, b, NULL
public, bogon, 16473, e, NULL
public, bogon2, 16480, e, NULL
public, bogon3, 16488, e, NULL
public, colors, 16496, e, NULL
public, empty_enum, 16518, e, NULL
public, huge_label, 16520, e, NULL
public, paint_color, 16510, e, NULL
public, planets, 16432, e, NULL
public, rainbow, 16418, e, NULL
schema1, enum_in_schema, 16527, c, c
schema2, enum_in_schema, 16530, c, c
```

Reviewers: xCluster, jhe

Reviewed By: jhe

Subscribers: ybase, yql

Differential Revision: https://phorge.dev.yugabyte.com/D42819

Author: mdbridge

src/postgres/src/backend/catalog/pg_type.c

@@ -39,6 +39,7 @@
 
 /* YB includes */
 #include "catalog/catalog.h"
+#include "catalog/yb_oid_assignment.h"
 #include "pg_yb_utils.h"
 
 static char *makeUniqueTypeName(const char *typeName, Oid typeNamespace,
@@ -142,6 +143,11 @@ TypeShellMake(const char *typeName, Oid typeNamespace, Oid ownerId)
 		typoid = binary_upgrade_next_pg_type_oid;
 		binary_upgrade_next_pg_type_oid = InvalidOid;
 	}
+	else if (YbUsingTypeOidAssignment())
+	{
+		typoid = YbLookupOidAssignmentForType(get_namespace_name(typeNamespace),
+											  typeName);
+	}
 	else
 	{
 		typoid = GetNewOidWithIndex(pg_type_desc, TypeOidIndexId,
@@ -524,6 +530,11 @@ TypeCreate(Oid newTypeOid,
 					(errcode(ERRCODE_INVALID_TABLE_DEFINITION),
 					 errmsg("system relations must have an explicit type OID!")));
 		}
+		else if (YbUsingTypeOidAssignment())
+		{
+			typeObjectId = YbLookupOidAssignmentForType(get_namespace_name(typeNamespace),
+														typeName);
+		}
 		/* else allow system to assign oid */
 		else
 		{

src/postgres/src/backend/catalog/yb_catalog/yb_oid_assignment.c

@@ -21,7 +21,6 @@
  *--------------------------------------------------------------------------------------------------
  */
 
-
 #include "postgres.h"
 
 #include "utils/builtins.h"
@@ -32,7 +31,8 @@
 static HTAB *yb_enum_label_assignment_map = NULL;
 static bool yb_enum_label_assignment_exists = false;
 
-static HTAB *yb_sequence_oid_assignment_map = NULL;
+static HTAB *yb_oid_assignment_map = NULL;
+static bool yb_type_oid_assignment_exists = false;
 static bool yb_sequence_oid_assignment_exists = false;
 
 /*
@@ -66,12 +66,6 @@ YbClearEnumLabelMap(void)
 static void
 YbCreateEnumLabelMapKey(Oid enum_oid, const char *label, char *key_buffer)
 {
-	/*
-	 * For now ignore enum OID field.  See YbLookupOidAssignmentForEnumLabel
-	 * for why.
-	 */
-	enum_oid = 42;
-
 	int written_bytes = snprintf(key_buffer,
 								 YB_ENUM_LABEL_ASSIGNMENT_MAP_KEY_SIZE, "%u.%s",
 								 enum_oid, label);
@@ -114,74 +108,78 @@ YbLookupOidForEnumLabel(Oid enum_oid, const char *label)
 	return InvalidOid;
 }
 
+#define YB_OID_KIND_TYPE "type"
+#define YB_OID_KIND_SEQUENCE "sequence"
+
 /*
- * yb_sequence_oid_assignment_map key format is <schema>.<name>\0
+ * yb_oid_assignment_map key format is <oid_kind>.<schema>.<name>\0
+ * <oid_kind>\0 is guaranteed to fit in 20 characters.
  * <identifier>\0 is guaranteed to fit in NAMEDATALEN characters.
  */
-#define YB_SEQUENCE_OID_ASSIGNMENT_MAP_KEY_SIZE (NAMEDATALEN + NAMEDATALEN)
+#define YB_OID_ASSIGNMENT_MAP_KEY_SIZE (20 + NAMEDATALEN + NAMEDATALEN)
 
-typedef struct YbSequenceOidAssignmentMapEntry {
-	/* encodes schema, name */
-	char		key[YB_SEQUENCE_OID_ASSIGNMENT_MAP_KEY_SIZE];
+typedef struct YbOidAssignmentMapEntry {
+	/* encodes oid_kind, schema, name */
+	char		key[YB_OID_ASSIGNMENT_MAP_KEY_SIZE];
 	Oid			oid;
-} YbSequenceOidAssignmentMapEntry;
+} YbOidAssignmentMapEntry;
 
 static void
-YbClearSequenceOidMap(void)
+YbClearOidMap(void)
 {
 	HASHCTL ctl;
 
-	if (yb_sequence_oid_assignment_map != NULL)
-		hash_destroy(yb_sequence_oid_assignment_map);
+	if (yb_oid_assignment_map != NULL)
+		hash_destroy(yb_oid_assignment_map);
 	memset(&ctl, 0, sizeof(ctl));
-	ctl.keysize = YB_SEQUENCE_OID_ASSIGNMENT_MAP_KEY_SIZE;
-	ctl.entrysize = sizeof(YbSequenceOidAssignmentMapEntry);
-	yb_sequence_oid_assignment_map = hash_create("YB sequence OIDs map",
-											   /* initial size */ 20, &ctl,
-											   HASH_ELEM | HASH_STRINGS);
+	ctl.keysize = YB_OID_ASSIGNMENT_MAP_KEY_SIZE;
+	ctl.entrysize = sizeof(YbOidAssignmentMapEntry);
+	yb_oid_assignment_map = hash_create("YB OIDs map",
+										/* initial size */ 20, &ctl,
+										HASH_ELEM | HASH_STRINGS);
 }
 
 static void
-YbCreateSequenceOidMapKey(const char *schema, const char *name, char *key_buffer)
+YbCreateOidMapKey(const char *oid_kind, const char *schema, const char *name,
+				  char *key_buffer)
 {
-	int written_bytes = snprintf(key_buffer,
-								 YB_SEQUENCE_OID_ASSIGNMENT_MAP_KEY_SIZE,
-								 "%s.%s", schema, name);
-	if (written_bytes >= YB_SEQUENCE_OID_ASSIGNMENT_MAP_KEY_SIZE)
+	int written_bytes = snprintf(key_buffer, YB_OID_ASSIGNMENT_MAP_KEY_SIZE,
+								 "%s.%s.%s", oid_kind, schema, name);
+	if (written_bytes >= YB_OID_ASSIGNMENT_MAP_KEY_SIZE)
 		elog(ERROR,
 			 "unexpectedly large schema/name in OID assignment (schema '%s', "
 			 "name '%s')",
 			 schema, name);
 }
 
 static void
-YbInsertSequenceOid(const char *schema, const char *name, Oid sequence_oid)
+YbInsertOid(const char *oid_kind, const char *schema, const char *name, Oid oid)
 {
-	char key[YB_SEQUENCE_OID_ASSIGNMENT_MAP_KEY_SIZE];
-	YbCreateSequenceOidMapKey(schema, name, key);
+	char key[YB_OID_ASSIGNMENT_MAP_KEY_SIZE];
+	YbCreateOidMapKey(oid_kind, schema, name, key);
 
 	bool found;
-	YbSequenceOidAssignmentMapEntry *entry =
-		hash_search(yb_sequence_oid_assignment_map, key, HASH_ENTER, &found);
+	YbOidAssignmentMapEntry *entry =
+		hash_search(yb_oid_assignment_map, key, HASH_ENTER, &found);
 	if (!found)
-		entry->oid = sequence_oid;
-	else if (entry->oid != sequence_oid)
+		entry->oid = oid;
+	else if (entry->oid != oid)
 		elog(ERROR,
-			 "attempt to provide multiple OIDs for sequence %s.%s: %u vs "
+			 "attempt to provide multiple OIDs for %s %s.%s: %u vs "
 			 "%u",
-			 schema, name, entry->oid, sequence_oid);
+			 oid_kind, schema, name, entry->oid, oid);
 }
 
 /* Returns InvalidOid on not found. */
 static Oid
-YbLookupOidForSequence(const char *schema, const char *name)
+YbLookupOid(const char *oid_kind, const char *schema, const char *name)
 {
-	char key[YB_SEQUENCE_OID_ASSIGNMENT_MAP_KEY_SIZE];
-	YbCreateSequenceOidMapKey(schema, name, key);
+	char key[YB_OID_ASSIGNMENT_MAP_KEY_SIZE];
+	YbCreateOidMapKey(oid_kind, schema, name, key);
 
 	bool found;
-	YbSequenceOidAssignmentMapEntry *entry =
-		hash_search(yb_sequence_oid_assignment_map, key, HASH_FIND, &found);
+	YbOidAssignmentMapEntry *entry =
+		hash_search(yb_oid_assignment_map, key, HASH_FIND, &found);
 	if (found)
 		return entry->oid;
 	return InvalidOid;
@@ -201,6 +199,76 @@ YbGetOidFromText(const text *input)
 	return result;
 }
 
+static void
+YbExtractEnumLabelMap(text *json_text, char *map_key, bool *found)
+{
+	text       *map_json = json_get_value(json_text, map_key);
+	if (!map_json)
+	{
+		*found = false;
+		return;
+	}
+	*found = true;
+
+	int length = get_json_array_length(map_json);
+	for (int i = 0; i < length; i++)
+	{
+		text *label_info_entry = get_json_array_element(map_json, i);
+		char *label = text_to_cstring(json_get_denormalized_value(label_info_entry,
+																  "label"));
+		text *label_oid_text = json_get_value(label_info_entry,
+											  "label_oid");
+		text *enum_oid_text = json_get_value(label_info_entry,
+											 "enum_oid");
+
+		Oid label_oid = YbGetOidFromText(label_oid_text);
+		Oid enum_oid = YbGetOidFromText(enum_oid_text);
+		if (label_oid == InvalidOid || enum_oid == InvalidOid)
+		{
+			elog(ERROR,
+				 "invalid JSON passed to "
+				 "yb_xcluster_set_next_oid_assignments: '%s'",
+				 text_to_cstring(json_text));
+		}
+
+		YbInsertEnumLabel(enum_oid, label, label_oid);
+	}
+}
+
+static void
+YbExtractNameToOidMap(text *json_text, char *map_key, const char *oid_kind,
+					  bool *found)
+{
+	text       *map_json = json_get_value(json_text, map_key);
+	if (!map_json)
+	{
+		*found = false;
+		return;
+	}
+	*found = true;
+
+	int length = get_json_array_length(map_json);
+	for (int i = 0; i < length; i++)
+	{
+		text *type_info_entry = get_json_array_element(map_json, i);
+		char *schema = text_to_cstring(json_get_denormalized_value(type_info_entry,
+																   "schema"));
+		char *name = text_to_cstring(json_get_denormalized_value(type_info_entry,
+																 "name"));
+		text *oid_text = json_get_value(type_info_entry, "oid");
+		Oid oid = YbGetOidFromText(oid_text);
+		if (oid == InvalidOid)
+		{
+			elog(ERROR,
+				 "invalid JSON passed to "
+				 "yb_xcluster_set_next_oid_assignments: '%s'",
+				 text_to_cstring(json_text));
+		}
+
+		YbInsertOid(oid_kind, schema, name, oid);
+	}
+}
+
 PG_FUNCTION_INFO_V1(yb_xcluster_set_next_oid_assignments);
 
 /*
@@ -231,6 +299,24 @@ PG_FUNCTION_INFO_V1(yb_xcluster_set_next_oid_assignments);
  *
  * Example:
  *    SELECT pg_catalog.yb_xcluster_set_next_oid_assignments(
+ *       '{"type_info":['                                        ||
+ *            '{"schema":"public","name":"my_range","oid":16406}' ||
+ *            ']}');
+ *
+ * This indicates that the type named my_range in schema public should
+ * be assigned the OID 16406.
+ *
+ * The type_info key is optional; if it is present then all types created
+ * directly via CREATE TYPE until the assignment is changed are expected to be
+ * covered by the assignment.  In the example this means that if the DDL
+ * attempts to create a range not_my_range then an error will occur.  It is
+ * not an error if the DDL does not create all the types mentioned in the
+ * assignment.  Multi-ranges are not covered here as they are not directly
+ * created via CREATE TYPE.
+ *
+ *
+ * Example:
+ *    SELECT pg_catalog.yb_xcluster_set_next_oid_assignments(
  *       '{"sequence_info":['                                        ||
  *            '{"schema":"public","name":"my_sequence","oid":16406}' ||
  *            ']}');
@@ -263,62 +349,14 @@ yb_xcluster_set_next_oid_assignments(PG_FUNCTION_ARGS)
 	text *json_text = PG_GETARG_TEXT_P(0);
 
 	YbClearEnumLabelMap();
-	yb_enum_label_assignment_exists = false;
-	text *enum_label_info = json_get_value(json_text, "enum_label_info");
-	if (enum_label_info != NULL)
-	{
-		yb_enum_label_assignment_exists = true;
-		int length = get_json_array_length(enum_label_info);
-		for (int i = 0; i < length; i++)
-		{
-			text *label_info_entry = get_json_array_element(enum_label_info, i);
-			char *label = text_to_cstring(json_get_denormalized_value(label_info_entry, "label"));
-			text *label_oid_text = json_get_value(label_info_entry,
-												  "label_oid");
-			text *enum_oid_text = json_get_value(label_info_entry,
-												 "enum_oid");
-
-			Oid label_oid = YbGetOidFromText(label_oid_text);
-			Oid enum_oid = YbGetOidFromText(enum_oid_text);
-			if (label_oid == InvalidOid || enum_oid == InvalidOid)
-			{
-				elog(ERROR,
-					 "corrupted JSON passed to "
-					 "yb_xcluster_set_next_oid_assignments: '%s'",
-					 text_to_cstring(json_text));
-			}
-
-			YbInsertEnumLabel(enum_oid, label, label_oid);
-		}
-	}
+	YbExtractEnumLabelMap(json_text, "enum_label_info",
+						  &yb_enum_label_assignment_exists);
 
-	YbClearSequenceOidMap();
-	yb_sequence_oid_assignment_exists = false;
-	text *sequence_info = json_get_value(json_text, "sequence_info");
-	if (sequence_info != NULL)
-	{
-		yb_sequence_oid_assignment_exists = true;
-		int length = get_json_array_length(sequence_info);
-		for (int i = 0; i < length; i++)
-		{
-			text *sequence_info_entry = get_json_array_element(sequence_info, i);
-			char *schema = text_to_cstring(json_get_denormalized_value(sequence_info_entry,
-				"schema"));
-			char *name = text_to_cstring(json_get_denormalized_value(sequence_info_entry,
-				"name"));
-			text *oid_text = json_get_value(sequence_info_entry, "oid");
-			Oid sequence_oid = YbGetOidFromText(oid_text);
-			if (sequence_oid == InvalidOid)
-			{
-				elog(ERROR,
-					 "corrupted JSON passed to "
-					 "yb_xcluster_set_next_oid_assignments: '%s'",
-					 text_to_cstring(json_text));
-			}
-
-			YbInsertSequenceOid(schema, name, sequence_oid);
-		}
-	}
+	YbClearOidMap();
+	YbExtractNameToOidMap(json_text, "type_info", YB_OID_KIND_TYPE,
+						  &yb_type_oid_assignment_exists);
+	YbExtractNameToOidMap(json_text, "sequence_info", YB_OID_KIND_SEQUENCE,
+						  &yb_sequence_oid_assignment_exists);
 
 	PG_RETURN_VOID();
 }
@@ -332,24 +370,29 @@ YbUsingEnumLabelOidAssignment(void)
 Oid
 YbLookupOidAssignmentForEnumLabel(Oid enum_oid, const char *label)
 {
-	/*----------
-	 * Currently we do not ensure that enums have the same *pg_type*
-	 * OIDs.  We will fix that later, but in the meantime we take
-	 * advantage of the fact that we currently never have a
-	 * replicating DDL that refers to two different enums to ignore
-	 * the actual enum OID field when comparing labels.
-	 *
-	 * See YbCreateEnumLabelMapKey for the temporary code to ignore the enum
-	 * OID field.
-	 *----------
-	 */
 	Oid label_oid = YbLookupOidForEnumLabel(enum_oid, label);
 	if (label_oid == InvalidOid)
 		elog(ERROR, "no OID assignment for enum label %u.%s in OID assignment",
 			 enum_oid, label);
 	return label_oid;
 }
 
+bool
+YbUsingTypeOidAssignment(void)
+{
+	return yb_type_oid_assignment_exists;
+}
+
+Oid
+YbLookupOidAssignmentForType(const char *schema, const char *name)
+{
+	Oid type_oid = YbLookupOid(YB_OID_KIND_TYPE, schema, name);
+	if (type_oid == InvalidOid)
+		elog(ERROR, "no OID assignment for type %s.%s in OID assignment",
+			 schema, name);
+	return type_oid;
+}
+
 bool
 YbUsingSequenceOidAssignment(void)
 {
@@ -359,7 +402,7 @@ YbUsingSequenceOidAssignment(void)
 Oid
 YbLookupOidAssignmentForSequence(const char *schema, const char *name)
 {
-	Oid sequence_oid = YbLookupOidForSequence(schema, name);
+	Oid sequence_oid = YbLookupOid(YB_OID_KIND_SEQUENCE, schema, name);
 	if (sequence_oid == InvalidOid)
 		elog(ERROR, "no OID assignment for sequence %s.%s in OID assignment",
 			 schema, name);