Implementation of tag.

We implement tag in this commit. Tag can be attached to target database
object, which maybe database, user, schema, table etc. Tag has allowed
values, when we attach one tag to target object, we should specify this
tag's value which must be in allowed values array, otherwise will error
out.

To use it, we will illustrate some example as following:
CREATE TAG tag_example allowed_values '123', 'abc';
CREATE USER tag_user;
ALTER USER tag_user TAG (tag_example = '123');
ALTER TAG tag_example ADD ALLOWED_VALUES 'def';
ALTER TAG tag_example DROP ALLOWED_VALUES 'abc';

Authored-by: Zhang Wenchao zwcpostgres@gmail.com

Author: zhangwenchao-123

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;