Implementation of Tag.

src/backend/catalog/Makefile

@@ -95,9 +95,9 @@ CATALOG_HEADERS := \
 	pg_sequence.h pg_publication.h pg_publication_rel.h pg_subscription.h \
 	pg_subscription_rel.h gp_partition_template.h pg_task.h pg_task_run_history.h \
 	pg_profile.h pg_password_history.h pg_directory_table.h gp_storage_server.h \
+	gp_storage_user_mapping.h pg_tag.h pg_tag_description.h \
 	gp_matview_aux.h \
-	gp_matview_tables.h \
-	gp_storage_user_mapping.h
+	gp_matview_tables.h
 
 USE_INTERNAL_FTS_FOUND := $(if $(findstring USE_INTERNAL_FTS,$(CFLAGS)),true,false)
 

src/backend/catalog/aclchk.c

@@ -58,6 +58,7 @@
 #include "catalog/pg_statistic_ext.h"
 #include "catalog/pg_subscription.h"
 #include "catalog/pg_tablespace.h"
+#include "catalog/pg_tag.h"
 #include "catalog/pg_transform.h"
 #include "catalog/pg_ts_config.h"
 #include "catalog/pg_ts_dict.h"
@@ -3791,6 +3792,9 @@ aclcheck_error(AclResult aclerr, ObjectType objtype,
 					case OBJECT_EXTPROTOCOL:
 						msg = gettext_noop("permission denied for external protocol %s");
 						break;
+					case OBJECT_TAG:
+						msg = gettext_noop("permission denied for tag %s");
+						break;
 						/* these currently aren't used */
 					case OBJECT_ACCESS_METHOD:
 					case OBJECT_AMOP:
@@ -3902,6 +3906,9 @@ aclcheck_error(AclResult aclerr, ObjectType objtype,
 					case OBJECT_TABLE:
 						msg = gettext_noop("must be owner of table %s");
 						break;
+					case OBJECT_TAG:
+						msg = gettext_noop("must be owner of tag %s");
+						break;
 					case OBJECT_TYPE:
 						msg = gettext_noop("must be owner of type %s");
 						break;
@@ -5647,6 +5654,33 @@ pg_opfamily_ownercheck(Oid opf_oid, Oid roleid)
 	return has_privs_of_role(roleid, ownerId);
 }
 
+/*
+ * Ownership check for a tag (specified by OID).
+ */
+bool
+pg_tag_ownercheck(Oid tag_oid, Oid roleid)
+{
+	HeapTuple	tuple;
+	Oid			ownerId;
+
+	/* Superusers bypass all permission checking. */
+	if (superuser_arg(roleid))
+		return true;
+
+	tuple = SearchSysCache1(TAGOID, ObjectIdGetDatum(tag_oid));
+	if (!HeapTupleIsValid(tuple))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("tag with OID %u does not exist",
+						tag_oid)));
+
+	ownerId = ((Form_pg_tag) GETSTRUCT(tuple))->tagowner;
+
+	ReleaseSysCache(tuple);
+
+	return has_privs_of_role(roleid, ownerId);
+}
+
 /*
  * Ownership check for a text search dictionary (specified by OID).
  */

src/backend/catalog/catalog.c

@@ -44,6 +44,8 @@
 #include "catalog/pg_shseclabel.h"
 #include "catalog/pg_subscription.h"
 #include "catalog/pg_tablespace.h"
+#include "catalog/pg_tag.h"
+#include "catalog/pg_tag_description.h"
 #include "catalog/pg_task.h"
 #include "catalog/pg_task_run_history.h"
 #include "catalog/pg_type.h"
@@ -447,7 +449,10 @@ IsSharedRelation(Oid relationId)
 		relationId == StorageServerRelationId ||
 
 		relationId == ProfileRelationId ||
-		relationId == PasswordHistoryRelationId)
+		relationId == PasswordHistoryRelationId ||
+
+		relationId == TagRelationId ||
+		relationId == TagDescriptionRelationId)
 		return true;
 
 	/* These are their indexes */
@@ -501,7 +506,12 @@ IsSharedRelation(Oid relationId)
 		relationId == ProfileOidIndexId ||
 		relationId == ProfileVerifyFunctionIndexId ||
 		relationId == PasswordHistoryRolePasswordIndexId ||
-		relationId == PasswordHistoryRolePasswordsetatIndexId)
+		relationId == PasswordHistoryRolePasswordsetatIndexId ||
+		relationId == TagNameIndexId ||
+		relationId == TagOidIndexId ||
+		relationId == TagDescriptionIndexId ||
+		relationId == TagDescriptionTagidvalueIndexId ||
+		relationId == TagDescriptionOidIndexId)
 	{
 		return true;
 	}

src/backend/catalog/dependency.c

@@ -58,6 +58,8 @@
 #include "catalog/pg_statistic_ext.h"
 #include "catalog/pg_subscription.h"
 #include "catalog/pg_tablespace.h"
+#include "catalog/pg_tag.h"
+#include "catalog/pg_tag_description.h"
 #include "catalog/pg_task.h"
 #include "catalog/pg_transform.h"
 #include "catalog/pg_trigger.h"
@@ -78,6 +80,7 @@
 #include "commands/schemacmds.h"
 #include "commands/seclabel.h"
 #include "commands/sequence.h"
+#include "commands/tag.h"
 #include "commands/taskcmds.h"
 #include "commands/trigger.h"
 #include "commands/typecmds.h"
@@ -209,6 +212,8 @@ static const Oid object_classes[] = {
 	DirectoryTableRelationId,	/* OCLASS_DIRECTORY_TABLE */
 	StorageServerRelationId,	/* OCLASS_STORAGE_SERVER */
 	StorageUserMappingRelationId,	/* OCLASS_STORAGE_USER_MAPPING */
+	TagRelationId,				/* OCLASS_TAG */
+	TagDescriptionRelationId,	/* OCLASS_TAG_DESCRIPTION */
 	ExtprotocolRelationId,		/* OCLASS_EXTPROTOCOL */
 	GpMatviewAuxId,				/* OCLASS_MATVIEW_AUX */
 	TaskRelationId,				/* OCLASS_TASK */
@@ -1474,14 +1479,31 @@ doDeletion(const ObjectAddress *object, int flags)
 
 					Assert(object->objectSubId == 0);
 					index_drop(object->objectId, concurrent, concurrent_lock_mode);
+
+					/*
+					 * Delete tag description.
+					 */
+					DeleteTagDescriptions(MyDatabaseId,
+										  object->classId,
+										  object->objectId);
 				}
 				else
 				{
 					if (object->objectSubId != 0)
 						RemoveAttributeById(object->objectId,
 											object->objectSubId);
 					else
+					{
 						heap_drop_with_catalog(object->objectId);
+
+						/*
+						 * Delete tag description.
+						 */
+						DeleteTagDescriptions(MyDatabaseId,
+											  object->classId,
+											  object->objectId);
+					}
+
 				}
 
 				/*
@@ -1551,6 +1573,12 @@ doDeletion(const ObjectAddress *object, int flags)
 
 		case OCLASS_SCHEMA:
 			RemoveSchemaById(object->objectId);
+			/*
+			 * Delete tag description.
+			 */
+			DeleteTagDescriptions(MyDatabaseId,
+								  object->classId,
+								  object->objectId);
 			break;
 		case OCLASS_TASK:
 			RemoveTaskById(object->objectId);
@@ -1593,6 +1621,8 @@ doDeletion(const ObjectAddress *object, int flags)
 		case OCLASS_PASSWORDHISTORY:
 		case OCLASS_STORAGE_SERVER:
 		case OCLASS_STORAGE_USER_MAPPING:
+		case OCLASS_TAG:
+		case OCLASS_TAG_DESCRIPTION:
 			elog(ERROR, "global objects cannot be deleted by doDeletion");
 			break;
 
@@ -2996,6 +3026,12 @@ getObjectClass(const ObjectAddress *object)
 		case StorageUserMappingRelationId:
 			return OCLASS_STORAGE_USER_MAPPING;
 
+		case TagRelationId:
+			return OCLASS_TAG;
+
+		case TagDescriptionRelationId:
+			return OCLASS_TAG_DESCRIPTION;
+
 		default:
 		{
 			struct CustomObjectClass *coc;