diff --git a/src/backend/commands/resgroupcmds.c b/src/backend/commands/resgroupcmds.c
index 15d2ff24ba868128d4b3e26e20959b20ebc470c2..092360391735934e0cffdc54c630b04036677b9c 100644
--- a/src/backend/commands/resgroupcmds.c
+++ b/src/backend/commands/resgroupcmds.c
@@ -49,6 +49,7 @@
 #define RESGROUP_MAX_CPU_RATE_LIMIT	(100)
 
 #define RESGROUP_MIN_MEMORY_LIMIT	(1)
+#define RESGROUP_MAX_MEMORY_LIMIT	(100)
 
 #define RESGROUP_MIN_MEMORY_SHARED_QUOTA	(0)
 #define RESGROUP_MAX_MEMORY_SHARED_QUOTA	(100)
@@ -76,83 +77,46 @@ typedef void (*ResourceGroupCallback) (bool isCommit, void *arg);
  */
 typedef struct ResourceGroupCallbackItem
 {
-	struct ResourceGroupCallbackItem *next;
-	struct ResourceGroupCallbackItem *prev;
 	ResourceGroupCallback callback;
 	void *arg;
 } ResourceGroupCallbackItem;
 
-static ResourceGroupCallbackItem ResourceGroup_callbacks_head =
-{
-	&ResourceGroup_callbacks_head, &ResourceGroup_callbacks_head, NULL, NULL
-};
-
-static ResourceGroupCallbackItem *ResourceGroup_callbacks = &ResourceGroup_callbacks_head;
+static ResourceGroupCallbackItem ResourceGroup_callback = {NULL, NULL};
 
 static int str2Int(const char *str, const char *prop);
 static ResGroupLimitType getResgroupOptionType(const char* defname);
-static const char * getResgroupOptionName(ResGroupLimitType type);
-static void parseStmtOptions(CreateResourceGroupStmt *stmt, ResGroupOpts *options);
-static void validateCapabilities(Relation rel, Oid groupid, ResGroupOpts *options, bool newGroup);
+static ResGroupCap getResgroupOptionValue(DefElem *defel);
+static const char *getResgroupOptionName(ResGroupLimitType type);
+static void checkResgroupCapLimit(ResGroupLimitType type, ResGroupCap value);
+static void parseStmtOptions(CreateResourceGroupStmt *stmt, ResGroupCaps *caps);
+static void validateCapabilities(Relation rel, Oid groupid, ResGroupCaps *caps, bool newGroup);
 static void insertResgroupCapabilityEntry(Relation rel, Oid groupid, uint16 type, char *value);
-static void updateResgroupCapabilities(Oid groupid, const ResGroupCaps *resgroupCaps);
-static void insertResgroupCapabilities(Oid groupid, ResGroupOpts *options);
+static void updateResgroupCapabilityEntry(Relation rel,
+										  Oid groupId,
+										  ResGroupLimitType limitType,
+										  ResGroupCap value);
+static void insertResgroupCapabilities(Relation rel, Oid groupId, ResGroupCaps *caps);
 static void deleteResgroupCapabilities(Oid groupid);
-static void createResGroupCallback(bool isCommit, void *arg);
-static void dropResGroupCallback(bool isCommit, void *arg);
-static void alterResGroupCallback(bool isCommit, void *arg);
+static void checkAuthIdForDrop(Oid groupId);
+static void createResgroupCallback(bool isCommit, void *arg);
+static void dropResgroupCallback(bool isCommit, void *arg);
+static void alterResgroupCallback(bool isCommit, void *arg);
 static void registerResourceGroupCallback(ResourceGroupCallback callback, void *arg);
 
-/*
- * Register callback functions for resource group related operations.
- *
- * At transaction end, the callback occurs post-commit or post-abort, so the
- * callback functions can only do non-critical cleanup.
- */
-static void
-registerResourceGroupCallback(ResourceGroupCallback callback, void *arg)
-{
-	ResourceGroupCallbackItem *item;
-
-	item = (ResourceGroupCallbackItem *)
-		MemoryContextAlloc(TopMemoryContext,
-						   sizeof(ResourceGroupCallbackItem));
-	item->callback = callback;
-	item->arg = arg;
-
-	item->prev = ResourceGroup_callbacks->prev;
-	item->next = ResourceGroup_callbacks;
-	item->prev->next = item;
-	item->next->prev = item;
-}
-
 /*
  * Call resource group related callback functions at transaction end.
  *
- * On COMMIT, the callback functions are processed as FIFO.
- * On ABORT,  the callback functions are processed as LIFO.
- *
  * Note the callback functions would be removed as being processed.
  */
 void
 AtEOXact_ResGroup(bool isCommit)
 {
-	ResourceGroupCallbackItem *current =
-		isCommit ? ResourceGroup_callbacks->next : ResourceGroup_callbacks->prev;
-	while (current != ResourceGroup_callbacks)
-	{
-		ResourceGroupCallbackItem *tmp = isCommit? current->next : current->prev;
-
-		ResourceGroupCallback callback = current->callback;
-		void *arg =  current->arg;
-
-		current->prev->next = current->next;
-		current->next->prev = current->prev;
-		pfree(current);
-		current = tmp;
+	if (ResourceGroup_callback.callback == NULL)
+		return;
 
-		callback(isCommit, arg);
-	}
+	/* make sure callback function will not error out */
+	ResourceGroup_callback.callback(isCommit, ResourceGroup_callback.arg);
+	ResourceGroup_callback.callback = NULL;
 }
 
 /*
@@ -162,6 +126,7 @@ void
 CreateResourceGroup(CreateResourceGroupStmt *stmt)
 {
 	Relation	pg_resgroup_rel;
+	Relation	pg_resgroupcapability_rel;
 	TupleDesc	pg_resgroup_dsc;
 	ScanKeyData scankey;
 	SysScanDesc sscan;
@@ -169,7 +134,7 @@ CreateResourceGroup(CreateResourceGroupStmt *stmt)
 	Oid			groupid;
 	Datum		new_record[Natts_pg_resgroup];
 	bool		new_record_nulls[Natts_pg_resgroup];
-	ResGroupOpts options;
+	ResGroupCaps caps;
 	int			nResGroups;
 
 	/* Permission check - only superuser can create groups. */
@@ -187,15 +152,19 @@ CreateResourceGroup(CreateResourceGroupStmt *stmt)
 				(errcode(ERRCODE_RESERVED_NAME),
 				 errmsg("resource group name \"none\" is reserved")));
 
-	MemSet(&options, 0, sizeof(options));
-	parseStmtOptions(stmt, &options);
+	MemSet(&caps, 0, sizeof(caps));
+	parseStmtOptions(stmt, &caps);
 
 	/*
-	 * Grant ExclusiveLock to serialize concurrent 'CREATE RESOURCE GROUP'
+	 * both CREATE and ALTER resource group need check the sum of cpu_rate_limit
+	 * and memory_limit and make sure the sum don't exceed 100. To make it simple,
+	 * acquire AccessExclusiveLock lock on pg_resgroupcapability at the beginning
+	 * of CREATE and ALTER
 	 */
-	pg_resgroup_rel = heap_open(ResGroupRelationId, ExclusiveLock);
+	pg_resgroupcapability_rel = heap_open(ResGroupCapabilityRelationId, AccessExclusiveLock);
+	pg_resgroup_rel = heap_open(ResGroupRelationId, RowExclusiveLock);
 
-	/* Check if max_resource_group limit is reached */
+	/* Check if MaxResourceGroups limit is reached */
 	sscan = systable_beginscan(pg_resgroup_rel, ResGroupRsgnameIndexId, false,
 							   SnapshotNow, 0, NULL);
 	nResGroups = 0;
@@ -237,7 +206,7 @@ CreateResourceGroup(CreateResourceGroupStmt *stmt)
 	new_record[Anum_pg_resgroup_rsgname - 1] =
 		DirectFunctionCall1(namein, CStringGetDatum(stmt->name));
 
-	new_record[Anum_pg_resgroup_parent - 1] = Int64GetDatum(0);
+	new_record[Anum_pg_resgroup_parent - 1] = ObjectIdGetDatum(0);
 
 	pg_resgroup_dsc = RelationGetDescr(pg_resgroup_rel);
 	tuple = heap_form_tuple(pg_resgroup_dsc, new_record, new_record_nulls);
@@ -249,7 +218,8 @@ CreateResourceGroup(CreateResourceGroupStmt *stmt)
 	CatalogUpdateIndexes(pg_resgroup_rel, tuple);
 
 	/* process the WITH (...) list items */
-	insertResgroupCapabilities(groupid, &options);
+	validateCapabilities(pg_resgroupcapability_rel, groupid, &caps, true);
+	insertResgroupCapabilities(pg_resgroupcapability_rel, groupid, &caps);
 
 	/* Dispatch the statement to segments */
 	if (Gp_role == GP_ROLE_DISPATCH)
@@ -267,23 +237,23 @@ CreateResourceGroup(CreateResourceGroupStmt *stmt)
 	}
 
 	heap_close(pg_resgroup_rel, NoLock);
+	heap_close(pg_resgroupcapability_rel, NoLock);
 
 	/* Add this group into shared memory */
 	if (IsResGroupActivated())
 	{
 		Oid			*callbackArg;
 
-		AllocResGroupEntry(groupid, &options);
-
+		AllocResGroupEntry(groupid, &caps);
 
 		/* Argument of callback function should be allocated in heap region */
 		callbackArg = (Oid *)MemoryContextAlloc(TopMemoryContext, sizeof(Oid));
 		*callbackArg = groupid;
-		registerResourceGroupCallback(createResGroupCallback, (void *)callbackArg);
+		registerResourceGroupCallback(createResgroupCallback, (void *)callbackArg);
 
 		/* Create os dependent part for this resource group */
 		ResGroupOps_CreateGroup(groupid);
-		ResGroupOps_SetCpuRateLimit(groupid, options.cpuRateLimit);
+		ResGroupOps_SetCpuRateLimit(groupid, caps.cpuRateLimit);
 	}
 	else if (Gp_role == GP_ROLE_DISPATCH)
 		ereport(WARNING,
@@ -298,12 +268,9 @@ void
 DropResourceGroup(DropResourceGroupStmt *stmt)
 {
 	Relation	 pg_resgroup_rel;
-	Relation	 authIdRel;
 	HeapTuple	 tuple;
 	ScanKeyData	 scankey;
 	SysScanDesc	 sscan;
-	ScanKeyData	 authid_scankey;
-	SysScanDesc	 authid_scan;
 	Oid			 groupid;
 	Oid			*callbackArg;
 
@@ -354,23 +321,7 @@ DropResourceGroup(DropResourceGroupStmt *stmt)
 	/*
 	 * Check to see if any roles are in this resource group.
 	 */
-	authIdRel = heap_open(AuthIdRelationId, RowExclusiveLock);
-	ScanKeyInit(&authid_scankey,
-				Anum_pg_authid_rolresgroup,
-				BTEqualStrategyNumber, F_OIDEQ,
-				ObjectIdGetDatum(groupid));
-
-	authid_scan = systable_beginscan(authIdRel, AuthIdRolResGroupIndexId, true,
-									 SnapshotNow, 1, &authid_scankey);
-
-	if (HeapTupleIsValid(systable_getnext(authid_scan)))
-		ereport(ERROR,
-				(errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
-				 errmsg("resource group \"%s\" is used by at least one role",
-						stmt->name)));
-
-	systable_endscan(authid_scan);
-	heap_close(authIdRel, RowExclusiveLock);
+	checkAuthIdForDrop(groupid);
 
 	/*
 	 * Delete the resource group from the catalog.
@@ -387,11 +338,11 @@ DropResourceGroup(DropResourceGroupStmt *stmt)
 	 */
 	DeleteSharedComments(groupid, ResGroupRelationId);
 
-	/* metadata tracking */
-	MetaTrackDropObject(ResGroupRelationId, groupid);
-
 	if (Gp_role == GP_ROLE_DISPATCH)
 	{
+		/* metadata tracking */
+		MetaTrackDropObject(ResGroupRelationId, groupid);
+
 		CdbDispatchUtilityStatement((Node *) stmt,
 									DF_CANCEL_ON_ERROR|
 									DF_WITH_SNAPSHOT|
@@ -405,7 +356,7 @@ DropResourceGroup(DropResourceGroupStmt *stmt)
 		/* Argument of callback function should be allocated in heap region */
 		callbackArg = (Oid *)MemoryContextAlloc(TopMemoryContext, sizeof(Oid));
 		*callbackArg = groupid;
-		registerResourceGroupCallback(dropResGroupCallback, (void *)callbackArg);
+		registerResourceGroupCallback(dropResgroupCallback, (void *)callbackArg);
 	}
 }
 
@@ -415,22 +366,14 @@ DropResourceGroup(DropResourceGroupStmt *stmt)
 void
 AlterResourceGroup(AlterResourceGroupStmt *stmt)
 {
-	Relation	pg_resgroup_rel;
 	Relation	pg_resgroupcapability_rel;
-	HeapTuple	tuple;
-	ScanKeyData	scankey;
-	SysScanDesc	sscan;
 	Oid			groupid;
-	ResourceGroupAlterCallbackContext *callbackCtx;
-	int			concurrency = -1;
-	int			cpuRateLimitNew = -1;
-	int			memSharedQuotaNew = -1;
-	int			memSpillRatioNew = -1;
-	int			memLimitNew = -1;
 	DefElem		*defel;
 	ResGroupLimitType	limitType;
 	ResGroupCaps		caps;
-	ResGroupOpts		opts;
+	ResGroupCap			*capArray;
+	ResGroupCap			value;
+	ResGroupCap			oldValue;
 
 	/* Permission check - only superuser can alter resource groups. */
 	if (!superuser())
@@ -443,112 +386,27 @@ AlterResourceGroup(AlterResourceGroupStmt *stmt)
 	defel = (DefElem *) lfirst(list_head(stmt->options));
 
 	limitType = getResgroupOptionType(defel->defname);
+	if (limitType == RESGROUP_LIMIT_TYPE_UNKNOWN)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("option \"%s\" not recognized", defel->defname)));
 
-	switch (limitType)
-	{
-		case RESGROUP_LIMIT_TYPE_CONCURRENCY:
-			concurrency = defGetInt64(defel);
-			if (concurrency < RESGROUP_MIN_CONCURRENCY)
-				ereport(ERROR,
-						(errcode(ERRCODE_INVALID_LIMIT_VALUE),
-						 errmsg("concurrency limit cannot be less than %d",
-								RESGROUP_MIN_CONCURRENCY)));
-			if (concurrency > RESGROUP_MAX_CONCURRENCY)
-				ereport(ERROR,
-						(errcode(ERRCODE_INVALID_LIMIT_VALUE),
-						 errmsg("concurrency limit cannot be greater than 'max_connections'")));
-			break;
-
-		case RESGROUP_LIMIT_TYPE_CPU:
-			cpuRateLimitNew = defGetInt64(defel);
-			if (cpuRateLimitNew < RESGROUP_MIN_CPU_RATE_LIMIT)
-				ereport(ERROR,
-						(errcode(ERRCODE_INVALID_LIMIT_VALUE),
-						 errmsg("cpu rate limit cannot be less than %d",
-								RESGROUP_MIN_CPU_RATE_LIMIT)));
-			if (cpuRateLimitNew > RESGROUP_MAX_CPU_RATE_LIMIT)
-				ereport(ERROR,
-						(errcode(ERRCODE_INVALID_LIMIT_VALUE),
-						 errmsg("cpu rate limit cannot be greater than %d",
-								RESGROUP_MAX_CPU_RATE_LIMIT)));
-			/* overall limit will be verified later after groupid is known */
-			break;
-
-		case RESGROUP_LIMIT_TYPE_MEMORY_SHARED_QUOTA:
-			memSharedQuotaNew = defGetInt64(defel);
-			if (memSharedQuotaNew < RESGROUP_MIN_MEMORY_SHARED_QUOTA)
-				ereport(ERROR,
-						(errcode(ERRCODE_INVALID_LIMIT_VALUE),
-						 errmsg("memory shared quota cannot be less than %d",
-								RESGROUP_MIN_MEMORY_SHARED_QUOTA)));
-			if (memSharedQuotaNew > RESGROUP_MAX_MEMORY_SHARED_QUOTA)
-				ereport(ERROR,
-						(errcode(ERRCODE_INVALID_LIMIT_VALUE),
-						 errmsg("memory shared quota cannot be greater than %d",
-								RESGROUP_MAX_MEMORY_SHARED_QUOTA)));
-			break;
-
-		case RESGROUP_LIMIT_TYPE_MEMORY:
-			memLimitNew = defGetInt64(defel);
-			if (memLimitNew < RESGROUP_MIN_MEMORY_LIMIT)
-				ereport(ERROR,
-						(errcode(ERRCODE_INVALID_LIMIT_VALUE),
-						 errmsg("memory limit cannot be less than %d",
-								RESGROUP_MIN_MEMORY_LIMIT)));
-			if (memLimitNew > RESGROUP_MAX_MEMORY_LIMIT)
-				ereport(ERROR,
-						(errcode(ERRCODE_INVALID_LIMIT_VALUE),
-						 errmsg("memory limit cannot be greater than %d",
-								RESGROUP_MAX_MEMORY_LIMIT)));
-			/* overall limit will be verified later after groupid is known */
-			break;
-		case RESGROUP_LIMIT_TYPE_MEMORY_SPILL_RATIO:
-			memSpillRatioNew = defGetInt64(defel);
-			if (memSpillRatioNew < RESGROUP_MIN_MEMORY_SPILL_RATIO)
-				ereport(ERROR,
-						(errcode(ERRCODE_INVALID_LIMIT_VALUE),
-						 errmsg("memory spill ratio cannot be less than %d",
-								RESGROUP_MIN_MEMORY_SPILL_RATIO)));
-			if (memSpillRatioNew > RESGROUP_MAX_MEMORY_SPILL_RATIO)
-				ereport(ERROR,
-						(errcode(ERRCODE_INVALID_LIMIT_VALUE),
-						 errmsg("memory spill ratio cannot be greater than %d",
-								RESGROUP_MAX_MEMORY_SPILL_RATIO)));
-			break;
-
-		default:
-			ereport(ERROR,
-					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-					 errmsg("unsupported resource group limit type '%s'", defel->defname)));
-	}
+	value = getResgroupOptionValue(defel);
+	checkResgroupCapLimit(limitType, value);
 
 	/*
 	 * Check the pg_resgroup relation to be certain the resource group already
 	 * exists.
 	 */
-	pg_resgroup_rel = heap_open(ResGroupRelationId, RowExclusiveLock);
-
-	ScanKeyInit(&scankey,
-				Anum_pg_resgroup_rsgname,
-				BTEqualStrategyNumber, F_NAMEEQ,
-				CStringGetDatum(stmt->name));
-
-	sscan = systable_beginscan(pg_resgroup_rel, ResGroupRsgnameIndexId, true,
-							   SnapshotNow, 1, &scankey);
-
-	tuple = systable_getnext(sscan);
-	if (!HeapTupleIsValid(tuple))
+	groupid = GetResGroupIdForName(stmt->name, RowExclusiveLock);
+	if (groupid == InvalidOid)
 		ereport(ERROR,
 				(errcode(ERRCODE_UNDEFINED_OBJECT),
 				 errmsg("resource group \"%s\" does not exist",
 						stmt->name)));
 
-	groupid = HeapTupleGetOid(tuple);
-	systable_endscan(sscan);
-	heap_close(pg_resgroup_rel, NoLock);
-
 	if (limitType == RESGROUP_LIMIT_TYPE_CONCURRENCY &&
-		concurrency == 0 &&
+		value == 0 &&
 		groupid == ADMINRESGROUP_OID)
 	{
 		ereport(ERROR,
@@ -556,12 +414,6 @@ AlterResourceGroup(AlterResourceGroupStmt *stmt)
 				 errmsg("admin_group must have at least one concurrency")));
 	}
 
-	/* Argument of callback function should be allocated in heap region */
-	callbackCtx = (ResourceGroupAlterCallbackContext *)
-		MemoryContextAlloc(TopMemoryContext, sizeof(*callbackCtx));
-	callbackCtx->groupid = groupid;
-	callbackCtx->limittype = limitType;
-
 	/*
 	 * In validateCapabilities() we scan all the resource groups
 	 * to check whether the total cpu_rate_limit exceed 100 or not.
@@ -571,64 +423,30 @@ AlterResourceGroup(AlterResourceGroupStmt *stmt)
 	pg_resgroupcapability_rel = heap_open(ResGroupCapabilityRelationId,
 										  AccessExclusiveLock);
 
-	/* Load currency resource group capabilities */
-	GetResGroupCapabilities(groupid, &caps);
-
-	/* Pick up the effective settings from caps */
-	ResGroupCapsToOpts(&caps, &opts);
-
-	/* Attempt to pick previous 'proposed' as 'value' */
-	ResGroupDecideConcurrencyCaps(groupid, &caps, &opts);
-	ResGroupDecideMemoryCaps(groupid, &caps, &opts);
-
-	switch (limitType)
-	{
-		case RESGROUP_LIMIT_TYPE_CONCURRENCY:
-			opts.concurrency = concurrency;
-			ResGroupDecideConcurrencyCaps(groupid, &caps, &opts);
-			break;
-
-		case RESGROUP_LIMIT_TYPE_CPU:
-			opts.cpuRateLimit = cpuRateLimitNew;
-			if (caps.cpuRateLimit.proposed < cpuRateLimitNew)
-				validateCapabilities(pg_resgroupcapability_rel,
-									 groupid, &opts, false);
-
-			caps.cpuRateLimit.value = cpuRateLimitNew;
-			caps.cpuRateLimit.proposed = cpuRateLimitNew;
-			break;
-
-		case RESGROUP_LIMIT_TYPE_MEMORY_SHARED_QUOTA:
-			opts.memSharedQuota = memSharedQuotaNew;
-
-			ResGroupDecideMemoryCaps(groupid, &caps, &opts);
-			break;
+	/* Load current resource group capabilities */
+	GetResGroupCapabilities(pg_resgroupcapability_rel, groupid, &caps);
 
-		case RESGROUP_LIMIT_TYPE_MEMORY:
-			opts.memLimit = memLimitNew;
-			if (caps.memLimit.proposed < memLimitNew)
-				validateCapabilities(pg_resgroupcapability_rel,
-									 groupid, &opts, false);
+	capArray = (ResGroupCap *) &caps;
+	oldValue = capArray[limitType];
+	capArray[limitType] = value;
 
-			ResGroupDecideMemoryCaps(groupid, &caps, &opts);
-			break;
-		case RESGROUP_LIMIT_TYPE_MEMORY_SPILL_RATIO:
-			caps.memSpillRatio.value = memSpillRatioNew;
-			caps.memSpillRatio.proposed = memSpillRatioNew;
-			break;
+	if ((limitType == RESGROUP_LIMIT_TYPE_CPU ||
+		 limitType == RESGROUP_LIMIT_TYPE_MEMORY) &&
+		oldValue < value)
+		validateCapabilities(pg_resgroupcapability_rel, groupid, &caps, false);
 
-		default:
-			ereport(ERROR,
-					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-					 errmsg("unsupported resource group limit type '%s'", defel->defname)));
-	}
-
-	updateResgroupCapabilities(groupid, &caps);
+	updateResgroupCapabilityEntry(pg_resgroupcapability_rel,
+								  groupid, limitType, value);
 
 	heap_close(pg_resgroupcapability_rel, NoLock);
 
 	if (Gp_role == GP_ROLE_DISPATCH)
 	{
+		MetaTrackUpdObject(ResGroupCapabilityRelationId,
+						   groupid,
+						   GetUserId(),
+						   "ALTER", defel->defname);
+
 		CdbDispatchUtilityStatement((Node *) stmt,
 									DF_CANCEL_ON_ERROR|
 									DF_WITH_SNAPSHOT|
@@ -637,13 +455,17 @@ AlterResourceGroup(AlterResourceGroupStmt *stmt)
 									NULL);
 	}
 
-	/* Bump command counter to make this change visible in the callback function alterResGroupCommitCallback() */
-	CommandCounterIncrement();
-
 	if (IsResGroupActivated())
 	{
+		ResourceGroupAlterCallbackContext *callbackCtx;
+
+		/* Argument of callback function should be allocated in heap region */
+		callbackCtx = (ResourceGroupAlterCallbackContext *)
+			MemoryContextAlloc(TopMemoryContext, sizeof(*callbackCtx));
+		callbackCtx->groupid = groupid;
+		callbackCtx->limittype = limitType;
 		callbackCtx->caps = caps;
-		registerResourceGroupCallback(alterResGroupCallback, (void *)callbackCtx);
+		registerResourceGroupCallback(alterResgroupCallback, (void *)callbackCtx);
 	}
 }
 
@@ -651,20 +473,20 @@ AlterResourceGroup(AlterResourceGroupStmt *stmt)
  * Get all the capabilities of one resource group in pg_resgroupcapability.
  */
 void
-GetResGroupCapabilities(Oid groupId, ResGroupCaps *resgroupCaps)
+GetResGroupCapabilities(Relation rel, Oid groupId, ResGroupCaps *resgroupCaps)
 {
 	SysScanDesc	sscan;
 	ScanKeyData	key;
 	HeapTuple	tuple;
 	bool isNull;
-	Relation	relResGroupCapability;
+
 	/*
 	 * By converting caps from (ResGroupCaps *) to an array of (ResGroupCap *)
 	 * we can access the individual capability via index, so we don't need
 	 * to use a switch case when setting them.
 	 */
-	ResGroupCap *caps = (ResGroupCap *) resgroupCaps;
-	ResourceOwner owner = NULL;
+	ResGroupCap *capArray = (ResGroupCap *) resgroupCaps;
+
 	/*
 	 * We maintain a bit mask to track which resgroup limit capability types
 	 * have been retrieved, when mask is 0 then no limit capability is found
@@ -672,22 +494,14 @@ GetResGroupCapabilities(Oid groupId, ResGroupCaps *resgroupCaps)
 	 */
 	int			mask = 0;
 
-	MemSet(caps, 0, sizeof(ResGroupCaps));
-
-	if (CurrentResourceOwner == NULL)
-	{
-		owner = ResourceOwnerCreate(NULL, "getResgroupCapabilityEntry");
-		CurrentResourceOwner = owner;
-	}
-
-	relResGroupCapability = heap_open(ResGroupCapabilityRelationId, AccessShareLock);
+	MemSet(capArray, 0, sizeof(ResGroupCaps));
 
 	ScanKeyInit(&key,
 				Anum_pg_resgroupcapability_resgroupid,
 				BTEqualStrategyNumber, F_OIDEQ,
 				ObjectIdGetDatum(groupId));
 
-	sscan = systable_beginscan(relResGroupCapability,
+	sscan = systable_beginscan(rel,
 							   ResGroupCapabilityResgroupidIndexId,
 							   true,
 							   SnapshotNow, 1, &key);
@@ -696,13 +510,11 @@ GetResGroupCapabilities(Oid groupId, ResGroupCaps *resgroupCaps)
 	{
 		Datum				typeDatum;
 		ResGroupLimitType	type;
-		Datum				valueDatum;
 		Datum				proposedDatum;
-		char				*value;
 		char				*proposed;
 
 		typeDatum = heap_getattr(tuple, Anum_pg_resgroupcapability_reslimittype,
-								 relResGroupCapability->rd_att, &isNull);
+								 rel->rd_att, &isNull);
 		type = (ResGroupLimitType) DatumGetInt16(typeDatum);
 
 		Assert(type > RESGROUP_LIMIT_TYPE_UNKNOWN);
@@ -711,29 +523,14 @@ GetResGroupCapabilities(Oid groupId, ResGroupCaps *resgroupCaps)
 
 		mask |= 1 << type;
 
-		valueDatum = heap_getattr(tuple, Anum_pg_resgroupcapability_value, relResGroupCapability->rd_att, &isNull);
-		value = TextDatumGetCString(valueDatum);
-		caps[type].value = str2Int(value, getResgroupOptionName(type));
-
-		proposedDatum = heap_getattr(tuple, Anum_pg_resgroupcapability_proposed, relResGroupCapability->rd_att, &isNull);
+		proposedDatum = heap_getattr(tuple, Anum_pg_resgroupcapability_proposed,
+									 rel->rd_att, &isNull);
 		proposed = TextDatumGetCString(proposedDatum);
-		caps[type].proposed = str2Int(proposed, getResgroupOptionName(type));
+		capArray[type] = str2Int(proposed, getResgroupOptionName(type));
 	}
 
 	systable_endscan(sscan);
 
-	/*
-	 * release lock here to guarantee we have no lock held when acquiring
-	 * resource group slot
-	 */
-	heap_close(relResGroupCapability, AccessShareLock);
-
-	if (owner)
-	{
-		CurrentResourceOwner = NULL;
-		ResourceOwnerDelete(owner);
-	}
-
 	if (!mask)
 	{
 		ereport(ERROR,
@@ -797,7 +594,8 @@ GetResGroupIdForRole(Oid roleid)
 	}
 
 	/* must access tuple before systable_endscan */
-	groupId = DatumGetObjectId(heap_getattr(tuple, Anum_pg_authid_rolresgroup, rel->rd_att, NULL));
+	groupId = DatumGetObjectId(heap_getattr(tuple, Anum_pg_authid_rolresgroup,
+							   rel->rd_att, NULL));
 
 	systable_endscan(sscan);
 
@@ -819,6 +617,83 @@ GetResGroupIdForRole(Oid roleid)
 	return groupId;
 }
 
+/*
+ * GetResGroupIdForName -- Return the Oid for a resource group name
+ *
+ * Notes:
+ *	Used by the various admin commands to convert a user supplied group name
+ *	to Oid.
+ */
+Oid
+GetResGroupIdForName(char *name, LOCKMODE lockmode)
+{
+	Relation	rel;
+	ScanKeyData scankey;
+	SysScanDesc scan;
+	HeapTuple	tuple;
+	Oid			rsgid;
+
+	rel = heap_open(ResGroupRelationId, lockmode);
+
+	/* SELECT oid FROM pg_resgroup WHERE rsgname = :1 */
+	ScanKeyInit(&scankey,
+				Anum_pg_resgroup_rsgname,
+				BTEqualStrategyNumber, F_NAMEEQ,
+				CStringGetDatum(name));
+	scan = systable_beginscan(rel, ResGroupRsgnameIndexId, true,
+							  SnapshotNow, 1, &scankey);
+
+	tuple = systable_getnext(scan);
+	if (HeapTupleIsValid(tuple))
+		rsgid = HeapTupleGetOid(tuple);
+	else
+		rsgid = InvalidOid;
+
+	systable_endscan(scan);
+	heap_close(rel, lockmode);
+
+	return rsgid;
+}
+
+/*
+ * GetResGroupNameForId -- Return the resource group name for an Oid
+ */
+char *
+GetResGroupNameForId(Oid oid, LOCKMODE lockmode)
+{
+	Relation	rel;
+	ScanKeyData scankey;
+	SysScanDesc scan;
+	HeapTuple	tuple;
+	char		*name = NULL;
+
+	rel = heap_open(ResGroupRelationId, lockmode);
+
+	/* SELECT rsgname FROM pg_resgroup WHERE oid = :1 */
+	ScanKeyInit(&scankey,
+				ObjectIdAttributeNumber,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(oid));
+	scan = systable_beginscan(rel, ResGroupOidIndexId, true,
+							  SnapshotNow, 1, &scankey);
+
+	tuple = systable_getnext(scan);
+	if (HeapTupleIsValid(tuple))
+	{
+		bool isnull;
+		Datum nameDatum = heap_getattr(tuple, Anum_pg_resgroup_rsgname,
+									   rel->rd_att, &isnull);
+		Assert (!isnull);
+		Name resGroupName = DatumGetName(nameDatum);
+		name = pstrdup(NameStr(*resGroupName));
+	}
+
+	systable_endscan(scan);
+	heap_close(rel, lockmode);
+
+	return name;
+}
+
 /*
  * Get the option type from a name string.
  *
@@ -843,6 +718,20 @@ getResgroupOptionType(const char* defname)
 		return RESGROUP_LIMIT_TYPE_UNKNOWN;
 }
 
+/*
+ * Get capability value from DefElem, convert from int64 to int
+ */
+static ResGroupCap
+getResgroupOptionValue(DefElem *defel)
+{
+	int64 value = defGetInt64(defel);
+	if (value < INT_MIN || value > INT_MAX)
+		ereport(ERROR,
+				(errcode(ERRCODE_SYNTAX_ERROR),
+				 errmsg("capability %s is out of range", defel->defname)));
+	return (ResGroupCap)value;
+}
+
 /*
  * Get the option name from type.
  *
@@ -871,41 +760,16 @@ getResgroupOptionName(ResGroupLimitType type)
 }
 
 /*
- * Parse a statement and store the settings in options.
- *
- * @param stmt     the statement
- * @param options  used to store the settings
+ * Check if capability value exceeds max and min value
  */
 static void
-parseStmtOptions(CreateResourceGroupStmt *stmt, ResGroupOpts *options)
+checkResgroupCapLimit(ResGroupLimitType type, int value)
 {
-	ListCell *cell;
-	int types = 0;
-
-	foreach(cell, stmt->options)
-	{
-		DefElem *defel = (DefElem *) lfirst(cell);
-		int type = getResgroupOptionType(defel->defname);
-
-		if (type == RESGROUP_LIMIT_TYPE_UNKNOWN)
-			ereport(ERROR,
-					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-					 errmsg("option \"%s\" not recognized", defel->defname)));
-
-		if (types & (1 << type))
-			ereport(ERROR,
-					(errcode(ERRCODE_SYNTAX_ERROR),
-					errmsg("Find duplicate resoure group resource type: %s",
-						   defel->defname)));
-		else
-			types |= 1 << type;
-
 		switch (type)
 		{
 			case RESGROUP_LIMIT_TYPE_CONCURRENCY:
-				options->concurrency = defGetInt64(defel);
-				if (options->concurrency < RESGROUP_MIN_CONCURRENCY ||
-					options->concurrency > RESGROUP_MAX_CONCURRENCY)
+				if (value < RESGROUP_MIN_CONCURRENCY ||
+					value > RESGROUP_MAX_CONCURRENCY)
 					ereport(ERROR,
 							(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 							errmsg("concurrency range is [%d, 'max_connections']",
@@ -913,9 +777,8 @@ parseStmtOptions(CreateResourceGroupStmt *stmt, ResGroupOpts *options)
 				break;
 
 			case RESGROUP_LIMIT_TYPE_CPU:
-				options->cpuRateLimit = defGetInt64(defel);
-				if (options->cpuRateLimit < RESGROUP_MIN_CPU_RATE_LIMIT ||
-					options->cpuRateLimit > RESGROUP_MAX_CPU_RATE_LIMIT)
+				if (value < RESGROUP_MIN_CPU_RATE_LIMIT ||
+					value > RESGROUP_MAX_CPU_RATE_LIMIT)
 					ereport(ERROR,
 							(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 							errmsg("cpu_rate_limit range is [%d, %d]",
@@ -924,9 +787,8 @@ parseStmtOptions(CreateResourceGroupStmt *stmt, ResGroupOpts *options)
 				break;
 
 			case RESGROUP_LIMIT_TYPE_MEMORY:
-				options->memLimit = defGetInt64(defel);
-				if (options->memLimit < RESGROUP_MIN_MEMORY_LIMIT ||
-					options->memLimit > RESGROUP_MAX_MEMORY_LIMIT)
+				if (value < RESGROUP_MIN_MEMORY_LIMIT ||
+					value > RESGROUP_MAX_MEMORY_LIMIT)
 					ereport(ERROR,
 							(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 							errmsg("memory_limit range is [%d, %d]",
@@ -935,9 +797,8 @@ parseStmtOptions(CreateResourceGroupStmt *stmt, ResGroupOpts *options)
 				break;
 
 			case RESGROUP_LIMIT_TYPE_MEMORY_SHARED_QUOTA:
-				options->memSharedQuota = defGetInt64(defel);
-				if (options->memSharedQuota < RESGROUP_MIN_MEMORY_SHARED_QUOTA ||
-					options->memSharedQuota > RESGROUP_MAX_MEMORY_SHARED_QUOTA)
+				if (value < RESGROUP_MIN_MEMORY_SHARED_QUOTA ||
+					value > RESGROUP_MAX_MEMORY_SHARED_QUOTA)
 					ereport(ERROR,
 							(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 							errmsg("memory_shared_quota range is [%d, %d]",
@@ -946,9 +807,8 @@ parseStmtOptions(CreateResourceGroupStmt *stmt, ResGroupOpts *options)
 				break;
 
 			case RESGROUP_LIMIT_TYPE_MEMORY_SPILL_RATIO:
-				options->memSpillRatio = defGetInt64(defel);
-				if (options->memSpillRatio < RESGROUP_MIN_MEMORY_SPILL_RATIO ||
-					options->memSpillRatio > RESGROUP_MAX_MEMORY_SPILL_RATIO)
+				if (value < RESGROUP_MIN_MEMORY_SPILL_RATIO ||
+					value > RESGROUP_MAX_MEMORY_SPILL_RATIO)
 					ereport(ERROR,
 							(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 							errmsg("memory_spill_ratio range is [%d, %d]",
@@ -960,21 +820,60 @@ parseStmtOptions(CreateResourceGroupStmt *stmt, ResGroupOpts *options)
 				Assert(!"unexpected options");
 				break;
 		}
+}
+
+/*
+ * Parse a statement and store the settings in options.
+ *
+ * @param stmt     the statement
+ * @param caps     used to store the settings
+ */
+static void
+parseStmtOptions(CreateResourceGroupStmt *stmt, ResGroupCaps *caps)
+{
+	ListCell *cell;
+	ResGroupCap value;
+	ResGroupCap *capArray = (ResGroupCap *)caps;
+	int mask = 0;
+
+	foreach(cell, stmt->options)
+	{
+		DefElem *defel = (DefElem *) lfirst(cell);
+		int type = getResgroupOptionType(defel->defname);
+
+		if (type == RESGROUP_LIMIT_TYPE_UNKNOWN)
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("option \"%s\" not recognized", defel->defname)));
+
+		if (mask & (1 << type))
+			ereport(ERROR,
+					(errcode(ERRCODE_SYNTAX_ERROR),
+					errmsg("Find duplicate resoure group resource type: %s",
+						   defel->defname)));
+		else
+			mask |= 1 << type;
+
+		value = getResgroupOptionValue(defel);
+		checkResgroupCapLimit(type, value);
+
+		capArray[type] = value;
 	}
 
-	if (options->memLimit == 0 || options->cpuRateLimit == 0)
+	if (!(mask & (1 << RESGROUP_LIMIT_TYPE_CPU)) ||
+		!(mask & (1 << RESGROUP_LIMIT_TYPE_MEMORY)))
 		ereport(ERROR,
 				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 				errmsg("must specify both memory_limit and cpu_rate_limit")));
 
-	if (!(types & (1 << RESGROUP_LIMIT_TYPE_CONCURRENCY)))
-		options->concurrency = RESGROUP_DEFAULT_CONCURRENCY;
+	if (!(mask & (1 << RESGROUP_LIMIT_TYPE_CONCURRENCY)))
+		caps->concurrency = RESGROUP_DEFAULT_CONCURRENCY;
 
-	if (!(types & (1 << RESGROUP_LIMIT_TYPE_MEMORY_SHARED_QUOTA)))
-		options->memSharedQuota = RESGROUP_DEFAULT_MEM_SHARED_QUOTA;
+	if (!(mask & (1 << RESGROUP_LIMIT_TYPE_MEMORY_SHARED_QUOTA)))
+		caps->memSharedQuota = RESGROUP_DEFAULT_MEM_SHARED_QUOTA;
 
-	if (!(types & (1 << RESGROUP_LIMIT_TYPE_MEMORY_SPILL_RATIO)))
-		options->memSpillRatio = RESGROUP_DEFAULT_MEM_SPILL_RATIO;
+	if (!(mask & (1 << RESGROUP_LIMIT_TYPE_MEMORY_SPILL_RATIO)))
+		caps->memSpillRatio = RESGROUP_DEFAULT_MEM_SPILL_RATIO;
 }
 
 /*
@@ -984,7 +883,7 @@ parseStmtOptions(CreateResourceGroupStmt *stmt, ResGroupOpts *options)
  * creates resource groups
  */
 static void
-createResGroupCallback(bool isCommit, void *arg)
+createResgroupCallback(bool isCommit, void *arg)
 {
 	Oid groupId;
 
@@ -1004,7 +903,7 @@ createResGroupCallback(bool isCommit, void *arg)
  * the queued transactions and cleanup shared menory entry.
  */
 static void
-dropResGroupCallback(bool isCommit, void *arg)
+dropResgroupCallback(bool isCommit, void *arg)
 {
 	Oid groupId;
 
@@ -1021,7 +920,7 @@ dropResGroupCallback(bool isCommit, void *arg)
  * transaction of this resource group may need to be woke up.
  */
 static void
-alterResGroupCallback(bool isCommit, void *arg)
+alterResgroupCallback(bool isCommit, void *arg)
 {
 	ResourceGroupAlterCallbackContext *ctx =
 		(ResourceGroupAlterCallbackContext *) arg;
@@ -1044,121 +943,95 @@ alterResGroupCallback(bool isCommit, void *arg)
  * handle the type conversion etc..
  *
  * @param groupid  oid of the resource group
- * @param options  the capabilities
+ * @param caps     the capabilities
  */
 static void
-insertResgroupCapabilities(Oid groupid,
-						   ResGroupOpts *options)
+insertResgroupCapabilities(Relation rel, Oid groupId, ResGroupCaps *caps)
 {
 	char value[64];
-	Relation resgroup_capability_rel = heap_open(ResGroupCapabilityRelationId, RowExclusiveLock);
-
-	validateCapabilities(resgroup_capability_rel, groupid, options, true);
 
-	sprintf(value, "%d", options->concurrency);
-	insertResgroupCapabilityEntry(resgroup_capability_rel, groupid, RESGROUP_LIMIT_TYPE_CONCURRENCY, value);
+	sprintf(value, "%d", caps->concurrency);
+	insertResgroupCapabilityEntry(rel, groupId,
+								  RESGROUP_LIMIT_TYPE_CONCURRENCY, value);
 
-	sprintf(value, "%d", options->cpuRateLimit);
-	insertResgroupCapabilityEntry(resgroup_capability_rel, groupid, RESGROUP_LIMIT_TYPE_CPU, value);
+	sprintf(value, "%d", caps->cpuRateLimit);
+	insertResgroupCapabilityEntry(rel, groupId,
+								  RESGROUP_LIMIT_TYPE_CPU, value);
 
-	sprintf(value, "%d", options->memLimit);
-	insertResgroupCapabilityEntry(resgroup_capability_rel, groupid, RESGROUP_LIMIT_TYPE_MEMORY, value);
+	sprintf(value, "%d", caps->memLimit);
+	insertResgroupCapabilityEntry(rel, groupId,
+								  RESGROUP_LIMIT_TYPE_MEMORY, value);
 
-	sprintf(value, "%d", options->memSharedQuota);
-	insertResgroupCapabilityEntry(resgroup_capability_rel, groupid, RESGROUP_LIMIT_TYPE_MEMORY_SHARED_QUOTA, value);
+	sprintf(value, "%d", caps->memSharedQuota);
+	insertResgroupCapabilityEntry(rel, groupId,
+								  RESGROUP_LIMIT_TYPE_MEMORY_SHARED_QUOTA, value);
 
-	sprintf(value, "%d", options->memSpillRatio);
-	insertResgroupCapabilityEntry(resgroup_capability_rel, groupid, RESGROUP_LIMIT_TYPE_MEMORY_SPILL_RATIO, value);
-
-	heap_close(resgroup_capability_rel, NoLock);
+	sprintf(value, "%d", caps->memSpillRatio);
+	insertResgroupCapabilityEntry(rel, groupId,
+								  RESGROUP_LIMIT_TYPE_MEMORY_SPILL_RATIO, value);
 }
 
 /*
  * Update all the capabilities of one resgroup in pg_resgroupcapability
  *
- * groupid is the update key, resgroupCaps contains all the capabilities.
+ * groupId and limitType are the scan keys.
  */
 static void
-updateResgroupCapabilities(Oid groupid, const ResGroupCaps *resgroupCaps)
+updateResgroupCapabilityEntry(Relation rel,
+							  Oid groupId,
+							  ResGroupLimitType limitType,
+							  ResGroupCap value)
 {
 	HeapTuple	oldTuple;
 	HeapTuple	newTuple;
 	SysScanDesc	sscan;
-	ScanKeyData	scankey;
+	ScanKeyData	scankey[2];
 	Datum		values[Natts_pg_resgroupcapability];
 	bool		isnull[Natts_pg_resgroupcapability];
 	bool		repl[Natts_pg_resgroupcapability];
-	/*
-	 * By converting caps from (ResGroupCaps *) to an array of (ResGroupCap *)
-	 * we can access the individual capability via index, so we don't need
-	 * to use a switch case when setting them.
-	 */
-	const ResGroupCap	*caps = (ResGroupCap *) resgroupCaps;
-	/*
-	 * We maintain a bit mask to track which resgroup limit capability types
-	 * have been retrieved, when mask is 0 then no limit capability is found
-	 * for the given groupid.
-	 */
-	int			mask = 0;
+	char		valueStr[16];
 
-	Relation resgroupCapabilityRel = heap_open(ResGroupCapabilityRelationId, RowExclusiveLock);
-
-	ScanKeyInit(&scankey,
+	ScanKeyInit(&scankey[0],
 				Anum_pg_resgroupcapability_resgroupid,
 				BTEqualStrategyNumber, F_OIDEQ,
-				ObjectIdGetDatum(groupid));
+				ObjectIdGetDatum(groupId));
+	ScanKeyInit(&scankey[1],
+				Anum_pg_resgroupcapability_reslimittype,
+				BTEqualStrategyNumber, F_INT2EQ,
+				Int16GetDatum(limitType));
 
-	sscan = systable_beginscan(resgroupCapabilityRel, ResGroupCapabilityResgroupidIndexId, true,
-							   SnapshotNow, 1, &scankey);
+	sscan = systable_beginscan(rel,
+							   ResGroupCapabilityResgroupidResLimittypeIndexId, true,
+							   SnapshotNow, 2, scankey);
 
 	MemSet(values, 0, sizeof(values));
 	MemSet(isnull, 0, sizeof(isnull));
 	MemSet(repl, 0, sizeof(repl));
 
-	while (HeapTupleIsValid(oldTuple = systable_getnext(sscan)))
-	{
-		Datum				typeDatum;
-		ResGroupLimitType	type;
-		char				value[16];
-		char				proposed[16];
-
-		typeDatum = heap_getattr(oldTuple, Anum_pg_resgroupcapability_reslimittype,
-								 resgroupCapabilityRel->rd_att, isnull);
-		type = (ResGroupLimitType) DatumGetInt16(typeDatum);
-
-		Assert(type > RESGROUP_LIMIT_TYPE_UNKNOWN);
-		Assert(type < RESGROUP_LIMIT_TYPE_COUNT);
-		Assert(!(mask & (1 << type)));
-
-		mask |= 1 << type;
+	oldTuple = systable_getnext(sscan);
+	if (!HeapTupleIsValid(oldTuple))
+		ereport(ERROR,
+				(errcode(ERRCODE_INTERNAL_ERROR),
+				 errmsg("capabilities missing for resource group %d type %d",
+						groupId, limitType)));
 
-		snprintf(value, sizeof(value), "%d", caps[type].value);
-		snprintf(proposed, sizeof(proposed), "%d", caps[type].proposed);
+	snprintf(valueStr, sizeof(valueStr), "%d", value);
 
-		values[Anum_pg_resgroupcapability_value - 1] = CStringGetTextDatum(value);
-		isnull[Anum_pg_resgroupcapability_value - 1] = false;
-		repl[Anum_pg_resgroupcapability_value - 1]  = true;
+	values[Anum_pg_resgroupcapability_value - 1] = CStringGetTextDatum(valueStr);
+	isnull[Anum_pg_resgroupcapability_value - 1] = false;
+	repl[Anum_pg_resgroupcapability_value - 1]  = true;
 
-		values[Anum_pg_resgroupcapability_proposed - 1] = CStringGetTextDatum(proposed);
-		isnull[Anum_pg_resgroupcapability_proposed - 1] = false;
-		repl[Anum_pg_resgroupcapability_proposed - 1]  = true;
+	values[Anum_pg_resgroupcapability_proposed - 1] = CStringGetTextDatum(valueStr);
+	isnull[Anum_pg_resgroupcapability_proposed - 1] = false;
+	repl[Anum_pg_resgroupcapability_proposed - 1]  = true;
 
-		newTuple = heap_modify_tuple(oldTuple, RelationGetDescr(resgroupCapabilityRel),
-									 values, isnull, repl);
+	newTuple = heap_modify_tuple(oldTuple, RelationGetDescr(rel),
+								 values, isnull, repl);
 
-		simple_heap_update(resgroupCapabilityRel, &oldTuple->t_self, newTuple);
-		CatalogUpdateIndexes(resgroupCapabilityRel, newTuple);
-	}
+	simple_heap_update(rel, &oldTuple->t_self, newTuple);
+	CatalogUpdateIndexes(rel, newTuple);
 
 	systable_endscan(sscan);
-	heap_close(resgroupCapabilityRel, NoLock);
-
-	if (!mask)
-	{
-		ereport(ERROR,
-				(errcode(ERRCODE_INTERNAL_ERROR),
-				 errmsg("capabilities missing for resource group: %d", groupid)));
-	}
 }
 
 /*
@@ -1171,20 +1044,21 @@ updateResgroupCapabilities(Oid groupid, const ResGroupCaps *resgroupCaps)
  *
  * @param rel      the relation
  * @param groupid  oid of the resource group
- * @param options  the options for the resource group
+ * @param caps     the capabilities for the resource group
  */
 static void
 validateCapabilities(Relation rel,
 					 Oid groupid,
-					 ResGroupOpts *options,
+					 ResGroupCaps *caps,
 					 bool newGroup)
 {
 	HeapTuple tuple;
 	SysScanDesc sscan;
-	int totalCpu = options->cpuRateLimit;
-	int totalMem = options->memLimit;
+	int totalCpu = caps->cpuRateLimit;
+	int totalMem = caps->memLimit;
 
-	sscan = systable_beginscan(rel, ResGroupCapabilityResgroupidIndexId, true, SnapshotNow, 0, NULL);
+	sscan = systable_beginscan(rel, ResGroupCapabilityResgroupidIndexId,
+							   true, SnapshotNow, 0, NULL);
 
 	while (HeapTupleIsValid(tuple = systable_getnext(sscan)))
 	{
@@ -1310,94 +1184,32 @@ deleteResgroupCapabilities(Oid groupid)
 }
 
 /*
- * GetResGroupIdForName -- Return the Oid for a resource group name
- *
- * Notes:
- *	Used by the various admin commands to convert a user supplied group name
- *	to Oid.
+ * Check to see if any roles are in this resource group.
  */
-Oid
-GetResGroupIdForName(char *name, LOCKMODE lockmode)
-{
-	Relation	rel;
-	ScanKeyData scankey;
-	SysScanDesc scan;
-	HeapTuple	tuple;
-	ResourceOwner owner = NULL;
-	Oid			rsgid;
-
-	if (CurrentResourceOwner == NULL)
-	{
-		owner = ResourceOwnerCreate(NULL, "GetResGroupIdForName");
-		CurrentResourceOwner = owner;
-	}
-
-	rel = heap_open(ResGroupRelationId, lockmode);
-
-	/* SELECT oid FROM pg_resgroup WHERE rsgname = :1 */
-	ScanKeyInit(&scankey,
-				Anum_pg_resgroup_rsgname,
-				BTEqualStrategyNumber, F_NAMEEQ,
-				CStringGetDatum(name));
-	scan = systable_beginscan(rel, ResGroupRsgnameIndexId, true,
-							  SnapshotNow, 1, &scankey);
-
-	tuple = systable_getnext(scan);
-	if (HeapTupleIsValid(tuple))
-		rsgid = HeapTupleGetOid(tuple);
-	else
-		rsgid = InvalidOid;
-
-	systable_endscan(scan);
-	heap_close(rel, lockmode);
-
-	if (owner)
-	{
-		CurrentResourceOwner = NULL;
-		ResourceOwnerDelete(owner);
-	}
-
-	return rsgid;
-}
-
-/*
- * GetResGroupNameForId -- Return the resource group name for an Oid
- */
-char *
-GetResGroupNameForId(Oid oid, LOCKMODE lockmode)
+static void
+checkAuthIdForDrop(Oid groupId)
 {
-	Relation	rel;
-	ScanKeyData scankey;
-	SysScanDesc scan;
-	HeapTuple	tuple;
-	char		*name = NULL;
-
-	rel = heap_open(ResGroupRelationId, lockmode);
+	Relation	 authIdRel;
+	ScanKeyData	 authidScankey;
+	SysScanDesc	 authidScan;
 
-	/* SELECT rsgname FROM pg_resgroup WHERE oid = :1 */
-	ScanKeyInit(&scankey,
-				ObjectIdAttributeNumber,
+	authIdRel = heap_open(AuthIdRelationId, RowExclusiveLock);
+	ScanKeyInit(&authidScankey,
+				Anum_pg_authid_rolresgroup,
 				BTEqualStrategyNumber, F_OIDEQ,
-				ObjectIdGetDatum(oid));
-	scan = systable_beginscan(rel, ResGroupOidIndexId, true,
-							  SnapshotNow, 1, &scankey);
+				ObjectIdGetDatum(groupId));
 
-	tuple = systable_getnext(scan);
-	if (HeapTupleIsValid(tuple))
-	{
-		bool isnull;
-		Datum nameDatum = heap_getattr(tuple, Anum_pg_resgroup_rsgname, rel->rd_att, &isnull);
-		Assert (!isnull);
-		Name resGroupName = DatumGetName(nameDatum);
-		name = pstrdup(NameStr(*resGroupName));
-	}
+	authidScan = systable_beginscan(authIdRel, AuthIdRolResGroupIndexId, true,
+									SnapshotNow, 1, &authidScankey);
 
-	systable_endscan(scan);
-	heap_close(rel, lockmode);
+	if (HeapTupleIsValid(systable_getnext(authidScan)))
+		ereport(ERROR,
+				(errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
+				 errmsg("resource group is used by at least one role")));
 
-	return name;
+	systable_endscan(authidScan);
+	heap_close(authIdRel, RowExclusiveLock);
 }
-
 /*
  * Convert a C str to a integer value.
  *
@@ -1420,3 +1232,13 @@ str2Int(const char *str, const char *prop)
 	return floor(val);
 }
 
+/*
+ * Register callback functions for resource group related operations.
+ */
+static void
+registerResourceGroupCallback(ResourceGroupCallback callback, void *arg)
+{
+	Assert(ResourceGroup_callback.callback == NULL);
+	ResourceGroup_callback.callback = callback;
+	ResourceGroup_callback.arg = arg;
+}
diff --git a/src/backend/utils/resgroup/resgroup.c b/src/backend/utils/resgroup/resgroup.c
index 0d5fb44717ccc7422a0a410ce575245c5e2f5c35..71487783aec7131f4dddb5dfa89812e69ed36329 100644
--- a/src/backend/utils/resgroup/resgroup.c
+++ b/src/backend/utils/resgroup/resgroup.c
@@ -391,15 +391,13 @@ error_out:
  * Allocate a resource group entry from a hash table
  */
 void
-AllocResGroupEntry(Oid groupId, const ResGroupOpts *opts)
+AllocResGroupEntry(Oid groupId, const ResGroupCaps *caps)
 {
 	ResGroupData	*group;
-	ResGroupCaps	caps;
 
 	LWLockAcquire(ResGroupLock, LW_EXCLUSIVE);
 
-	ResGroupOptsToCaps(opts, &caps);
-	group = createGroup(groupId, &caps);
+	group = createGroup(groupId, caps);
 	Assert(group != NULL);
 
 	LWLockRelease(ResGroupLock);
@@ -422,14 +420,11 @@ InitResGroups(void)
 	Relation			relResGroup;
 	Relation			relResGroupCapability;
 
+	on_shmem_exit(AtProcExit_ResGroup, 0);
+
 	/*
-	 * On master, the postmaster does the initialization
-	 * On segments, the first QE does the initialization
+	 * On master and segments, the first backend does the initialization.
 	 */
-	if (Gp_role == GP_ROLE_DISPATCH && GpIdentity.segindex != MASTER_CONTENT_ID)
-		return;
-
-	on_shmem_exit(AtProcExit_ResGroup, 0);
 	if (pResGroupControl->loaded)
 		return;
 	/*
@@ -480,8 +475,8 @@ InitResGroups(void)
 		int cpuRateLimit;
 		Oid groupId = HeapTupleGetOid(tuple);
 
-		GetResGroupCapabilities(groupId, &caps);
-		cpuRateLimit = caps.cpuRateLimit.value;
+		GetResGroupCapabilities(relResGroupCapability, groupId, &caps);
+		cpuRateLimit = caps.cpuRateLimit;
 
 		group = createGroup(groupId, &caps);
 		Assert(group != NULL);
@@ -499,6 +494,11 @@ InitResGroups(void)
 
 exit:
 	LWLockRelease(ResGroupLock);
+
+	/*
+	 * release lock here to guarantee we have no lock held when acquiring
+	 * resource group slot
+	 */
 	heap_close(relResGroup, AccessShareLock);
 	heap_close(relResGroupCapability, AccessShareLock);
 	CurrentResourceOwner = NULL;
@@ -654,9 +654,9 @@ ResGroupAlterOnCommit(Oid groupId,
 
 		if (limittype == RESGROUP_LIMIT_TYPE_CPU)
 		{
-			ResGroupOps_SetCpuRateLimit(groupId, caps->cpuRateLimit.proposed);
+			ResGroupOps_SetCpuRateLimit(groupId, caps->cpuRateLimit);
 		}
-		else
+		else if (limittype != RESGROUP_LIMIT_TYPE_MEMORY_SPILL_RATIO)
 		{
 			shouldWakeUp = groupApplyMemCaps(group, caps);
 
@@ -895,8 +895,8 @@ ResGroupReserveMemory(int32 memoryChunks, int32 overuseChunks, bool *waiverUsed)
 		selfUnassignDroppedGroup();
 		self->doMemCheck = false;
 
-		LOG_RESGROUP_DEBUG(LOG, "resource group is concurrently dropped while reserving memory: "
-						   "dropped group=%d, my group=%d",
+		LOG_RESGROUP_DEBUG(LOG, "resource group is concurrently dropped while "
+						   "reserving memory: dropped group=%d, my group=%d",
 						   groupGroupId, selfGroupId);
 
 		return true;
@@ -967,8 +967,8 @@ ResGroupReleaseMemory(int32 memoryChunks)
 		selfUnassignDroppedGroup();
 		self->doMemCheck = false;
 
-		LOG_RESGROUP_DEBUG(LOG, "resource group is concurrently dropped while releasing memory: "
-						   "dropped group=%d, my group=%d",
+		LOG_RESGROUP_DEBUG(LOG, "resource group is concurrently dropped while "
+						   "releasing memory: dropped group=%d, my group=%d",
 						   groupGroupId, selfGroupId);
 
 		return;
@@ -979,101 +979,6 @@ ResGroupReleaseMemory(int32 memoryChunks)
 	groupDecMemUsage(group, slot, memoryChunks);
 }
 
-/*
- * Decide the new resource group concurrency capabilities
- * of pg_resgroupcapability.
- *
- * The decision is based on current runtime information:
- * - 'proposed' will always be set to the latest setting;
- * - 'value' will be set to the most recent version of concurrency
- *   with which current nRunning doesn't exceed the limit;
- */
-void
-ResGroupDecideConcurrencyCaps(Oid groupId,
-							  ResGroupCaps *caps,
-							  const ResGroupOpts *opts)
-{
-	ResGroupData	*group;
-
-	/* If resource group is not in use we can always pick the new settings. */
-	if (!IsResGroupActivated())
-	{
-		caps->concurrency.value = opts->concurrency;
-		caps->concurrency.proposed = opts->concurrency;
-		return;
-	}
-
-	LWLockAcquire(ResGroupLock, LW_SHARED);
-
-	group = groupHashFind(groupId, true);
-
-	/*
-	 * If the runtime usage information doesn't exceed the new setting
-	 * then we can pick this setting as the new 'value'.
-	 */
-	if (group->nRunning <= opts->concurrency)
-		caps->concurrency.value = opts->concurrency;
-
-	/* 'proposed' is always set with latest setting */
-	caps->concurrency.proposed = opts->concurrency;
-
-	LWLockRelease(ResGroupLock);
-}
-
-/*
- * Decide the new resource group memory capabilities
- * of pg_resgroupcapability.
- *
- * The decision is based on current runtime information:
- * - 'proposed' will always be set to the latest setting;
- * - 'value' will be set to the most recent version of memory settings
- *   with which current memory quota usage and memory shared usage
- *   doesn't exceed the limit;
- */
-void
-ResGroupDecideMemoryCaps(int groupId,
-						 ResGroupCaps *caps,
-						 const ResGroupOpts *opts)
-{
-	ResGroupData	*group;
-	ResGroupCaps	capsNew;
-
-	/* If resource group is not in use we can always pick the new settings. */
-	if (!IsResGroupActivated())
-	{
-		caps->memLimit.value = opts->memLimit;
-		caps->memLimit.proposed = opts->memLimit;
-
-		caps->memSharedQuota.value = opts->memSharedQuota;
-		caps->memSharedQuota.proposed = opts->memSharedQuota;
-
-		return;
-	}
-
-	LWLockAcquire(ResGroupLock, LW_SHARED);
-
-	group = groupHashFind(groupId, true);
-
-	ResGroupOptsToCaps(opts, &capsNew);
-	/*
-	 * If the runtime usage information doesn't exceed the new settings
-	 * then we can pick these settings as the new 'value's.
-	 */
-	if (opts->memLimit <= caps->memLimit.proposed &&
-		group->memQuotaUsed <= groupGetMemQuotaExpected(&capsNew) &&
-		group->memSharedUsage <= groupGetMemSharedExpected(&capsNew))
-	{
-		caps->memLimit.value = opts->memLimit;
-		caps->memSharedQuota.value = opts->memSharedQuota;
-	}
-
-	/* 'proposed' is always set with latest setting */
-	caps->memSharedQuota.proposed = opts->memSharedQuota;
-	caps->memLimit.proposed = opts->memLimit;
-
-	LWLockRelease(ResGroupLock);
-}
-
 int64
 ResourceGroupGetQueryMemoryLimit(void)
 {
@@ -1350,7 +1255,7 @@ groupGetSlot(ResGroupData *group)
 	caps = &group->caps;
 
 	/* First check if the concurrency limit is reached */
-	if (group->nRunning >= caps->concurrency.proposed)
+	if (group->nRunning >= caps->concurrency)
 		return NULL;
 
 	if (!groupReserveMemQuota(group))
@@ -1632,14 +1537,14 @@ groupApplyMemCaps(ResGroupData *group, const ResGroupCaps *caps)
 	/* memQuotaAvailable is the total free non-shared quota */
 	memQuotaAvailable = group->memQuotaGranted - group->memQuotaUsed;
 
-	if (caps->concurrency.proposed > group->nRunning)
+	if (caps->concurrency > group->nRunning)
 	{
 		/*
 		 * memQuotaNeeded is the total non-shared quota needed
 		 * by all the free slots
 		 */
 		memQuotaNeeded = slotGetMemQuotaExpected(caps) *
-			(caps->concurrency.proposed - group->nRunning);
+			(caps->concurrency - group->nRunning);
 
 		/*
 		 * if memQuotaToFree > 0 then we can safely release these
@@ -1794,7 +1699,7 @@ static int32
 groupGetMemExpected(const ResGroupCaps *caps)
 {
 	Assert(pResGroupControl->totalChunks > 0);
-	return pResGroupControl->totalChunks * caps->memLimit.proposed / 100;
+	return pResGroupControl->totalChunks * caps->memLimit / 100;
 }
 
 /*
@@ -1803,11 +1708,11 @@ groupGetMemExpected(const ResGroupCaps *caps)
 static int32
 groupGetMemQuotaExpected(const ResGroupCaps *caps)
 {
-	if (caps->concurrency.proposed > 0)
-		return slotGetMemQuotaExpected(caps) * caps->concurrency.proposed;
+	if (caps->concurrency > 0)
+		return slotGetMemQuotaExpected(caps) * caps->concurrency;
 	else
 		return groupGetMemExpected(caps) *
-			(100 - caps->memSharedQuota.proposed) / 100;
+			(100 - caps->memSharedQuota) / 100;
 }
 
 /*
@@ -1834,10 +1739,10 @@ groupGetMemSpillTotal(const ResGroupCaps *caps)
 static int32
 slotGetMemQuotaExpected(const ResGroupCaps *caps)
 {
-	Assert(caps->concurrency.proposed != 0);
+	Assert(caps->concurrency != 0);
 	return groupGetMemExpected(caps) *
-		(100 - caps->memSharedQuota.proposed) / 100 /
-		caps->concurrency.proposed;
+		(100 - caps->memSharedQuota) / 100 /
+		caps->concurrency;
 }
 
 /*
@@ -1846,8 +1751,8 @@ slotGetMemQuotaExpected(const ResGroupCaps *caps)
 static int32
 slotGetMemSpill(const ResGroupCaps *caps)
 {
-	Assert(caps->concurrency.proposed != 0);
-	return groupGetMemSpillTotal(caps) / caps->concurrency.proposed;
+	Assert(caps->concurrency != 0);
+	return groupGetMemSpillTotal(caps) / caps->concurrency;
 }
 
 /*
@@ -1958,7 +1863,7 @@ mempoolAutoRelease(ResGroupData *group, ResGroupSlotData *slot)
 	memQuotaNeedFree = group->memQuotaGranted - groupGetMemQuotaExpected(caps);
 	memQuotaToFree = memQuotaNeedFree > 0 ? Min(memQuotaNeedFree, slot->memQuota) : 0;
 
-	if (caps->concurrency.proposed > 0)
+	if (caps->concurrency > 0)
 	{
 		/*
 		 * Under this situation, when this slot is released,
@@ -2015,8 +1920,12 @@ addTotalQueueDuration(ResGroupData *group)
 
 	TimestampTz start = pgstat_fetch_resgroup_queue_timestamp();
 	TimestampTz now = GetCurrentTimestamp();
-	Datum durationDatum = DirectFunctionCall2(timestamptz_age, TimestampTzGetDatum(now), TimestampTzGetDatum(start));
-	Datum sumDatum = DirectFunctionCall2(interval_pl, IntervalPGetDatum(&group->totalQueuedTime), durationDatum);
+	Datum durationDatum = DirectFunctionCall2(timestamptz_age,
+											  TimestampTzGetDatum(now),
+											  TimestampTzGetDatum(start));
+	Datum sumDatum = DirectFunctionCall2(interval_pl,
+										 IntervalPGetDatum(&group->totalQueuedTime),
+										 durationDatum);
 	memcpy(&group->totalQueuedTime, DatumGetIntervalP(sumDatum), sizeof(Interval));
 }
 
@@ -2068,11 +1977,8 @@ SerializeResGroupInfo(StringInfo str)
 
 	for (i = 0; i < RESGROUP_LIMIT_TYPE_COUNT; i++)
 	{
-		tmp = htonl(caps[i].value);
-		appendBinaryStringInfo(str, (char *) &tmp, sizeof(caps[i].value));
-
-		tmp = htonl(caps[i].proposed);
-		appendBinaryStringInfo(str, (char *) &tmp, sizeof(caps[i].proposed));
+		tmp = htonl(caps[i]);
+		appendBinaryStringInfo(str, (char *) &tmp, sizeof(caps[i]));
 	}
 }
 
@@ -2098,13 +2004,9 @@ DeserializeResGroupInfo(struct ResGroupCaps *capsOut,
 
 	for (i = 0; i < RESGROUP_LIMIT_TYPE_COUNT; i++)
 	{
-		memcpy(&tmp, ptr, sizeof(caps[i].value));
-		caps[i].value = ntohl(tmp);
-		ptr += sizeof(caps[i].value);
-
-		memcpy(&tmp, ptr, sizeof(caps[i].proposed));
-		caps[i].proposed = ntohl(tmp);
-		ptr += sizeof(caps[i].proposed);
+		memcpy(&tmp, ptr, sizeof(caps[i]));
+		caps[i] = ntohl(tmp);
+		ptr += sizeof(caps[i]);
 	}
 
 	Assert(len == ptr - buf);
@@ -2314,7 +2216,7 @@ SwitchResGroupOnSegment(const char *buf, int len)
 
 	/* Init self */
 	Assert(host_segments > 0);
-	Assert(caps.concurrency.proposed > 0);
+	Assert(caps.concurrency > 0);
 	selfSetGroup(group);
 	self->caps = caps;
 
@@ -2489,7 +2391,10 @@ groupHashRemove(Oid groupId)
 
 	Assert(LWLockHeldExclusiveByMe(ResGroupLock));
 
-	entry = (ResGroupHashEntry*)hash_search(pResGroupControl->htbl, (void *) &groupId, HASH_FIND, &found);
+	entry = (ResGroupHashEntry*)hash_search(pResGroupControl->htbl,
+											(void *) &groupId,
+											HASH_REMOVE,
+											&found);
 	if (!found)
 		ereport(ERROR,
 				(errcode(ERRCODE_DATA_CORRUPTED),
@@ -2502,8 +2407,6 @@ groupHashRemove(Oid groupId)
 	group->memSharedGranted = 0;
 	group->groupId = InvalidOid;
 
-	hash_search(pResGroupControl->htbl, (void *) &groupId, HASH_REMOVE, &found);
-
 	wakeupGroups(groupId);
 }
 
@@ -2595,8 +2498,9 @@ groupSetMemorySpillRatio(const ResGroupCaps *caps)
 {
 	char value[64];
 
-	snprintf(value, sizeof(value), "%d", caps->memSpillRatio.proposed);
-	set_config_option("memory_spill_ratio", value, PGC_USERSET, PGC_S_RESGROUP, GUC_ACTION_SET, true);
+	snprintf(value, sizeof(value), "%d", caps->memSpillRatio);
+	set_config_option("memory_spill_ratio", value, PGC_USERSET, PGC_S_RESGROUP,
+			GUC_ACTION_SET, true);
 }
 
 void
@@ -2605,41 +2509,10 @@ ResGroupGetMemInfo(int *memLimit, int *slotQuota, int *sharedQuota)
 	const ResGroupCaps *caps = &self->caps;
 
 	*memLimit = groupGetMemExpected(caps);
-	*slotQuota = caps->concurrency.proposed ? slotGetMemQuotaExpected(caps) : -1;
+	*slotQuota = caps->concurrency ? slotGetMemQuotaExpected(caps) : -1;
 	*sharedQuota = groupGetMemSharedExpected(caps);
 }
 
-/*
- * Convert ResGroupOpts to ResGroupCaps
- */
-void
-ResGroupOptsToCaps(const ResGroupOpts *optsIn, ResGroupCaps *capsOut)
-{
-	int i;
-	ResGroupCap		*caps = (ResGroupCap *) capsOut;
-	const int32		*opts = (int32 *) optsIn;
-
-	for (i = 0; i < RESGROUP_LIMIT_TYPE_COUNT; i++)
-	{
-		caps[i].value = opts[i];
-		caps[i].proposed = opts[i];
-	}
-}
-
-/*
- * Convert ResGroupCaps to ResGroupOpts
- */
-void
-ResGroupCapsToOpts(const ResGroupCaps *capsIn, ResGroupOpts *optsOut)
-{
-	int i;
-	const ResGroupCap	*caps = (ResGroupCap *) capsIn;
-	int32				*opts = (int32 *) optsOut;
-
-	for (i = 0; i < RESGROUP_LIMIT_TYPE_COUNT; i++)
-		opts[i] = caps[i].proposed;
-}
-
 /*
  * Validate the consistency of the resgroup information in self.
  *
@@ -3288,7 +3161,7 @@ resgroupDumpCaps(StringInfo str, ResGroupCap *caps)
 	appendStringInfo(str, "\"caps\":[");
 	for (i = 1; i < RESGROUP_LIMIT_TYPE_COUNT; i++)
 	{
-		appendStringInfo(str, "{\"value\":%d,\"proposed\":%d}", caps[i].value, caps[i].proposed);
+		appendStringInfo(str, "{\"%d\":%d}", i, caps[i]);
 		if (i < RESGROUP_LIMIT_TYPE_COUNT - 1)
 			appendStringInfo(str, ",");
 	}
diff --git a/src/bin/pg_dump/pg_dumpall.c b/src/bin/pg_dump/pg_dumpall.c
index e643d5b63fd183255bc3828cdd7ec5ebdbc972b9..4628ebfb3a6f284607ef9c50163f31a494b19841 100644
--- a/src/bin/pg_dump/pg_dumpall.c
+++ b/src/bin/pg_dump/pg_dumpall.c
@@ -636,11 +636,11 @@ dumpResGroups(PGconn *conn)
 			i_memory_spill_ratio;
 
 	printfPQExpBuffer(buf, "SELECT g.rsgname AS groupname, "
-					  "t1.proposed AS concurrency, "
-					  "t2.proposed AS cpu_rate_limit, "
-					  "t3.proposed AS memory_limit, "
-					  "t4.proposed AS memory_shared_quota, "
-					  "t5.proposed AS memory_spill_ratio "
+					  "t1.value AS concurrency, "
+					  "t2.value AS cpu_rate_limit, "
+					  "t3.value AS memory_limit, "
+					  "t4.value AS memory_shared_quota, "
+					  "t5.value AS memory_spill_ratio "
 					  "FROM pg_resgroup g, "
 					  "pg_resgroupcapability t1, "
 					  "pg_resgroupcapability t2, "
diff --git a/src/include/commands/resgroupcmds.h b/src/include/commands/resgroupcmds.h
index d1dfb726a00f89aad9808eea2895aa948079f03b..3bb7f83e9fb1d46cc6c79bcc7e5eae640ca11d5e 100644
--- a/src/include/commands/resgroupcmds.h
+++ b/src/include/commands/resgroupcmds.h
@@ -16,8 +16,7 @@
 
 #include "nodes/parsenodes.h"
 #include "utils/resgroup.h"
-
-#define RESGROUP_MAX_MEMORY_LIMIT	(100)
+#include "utils/relcache.h"
 
 extern void CreateResourceGroup(CreateResourceGroupStmt *stmt);
 extern void DropResourceGroup(DropResourceGroupStmt *stmt);
@@ -27,7 +26,9 @@ extern void AlterResourceGroup(AlterResourceGroupStmt *stmt);
 extern Oid GetResGroupIdForName(char *name, LOCKMODE lockmode);
 extern char *GetResGroupNameForId(Oid oid, LOCKMODE lockmode);
 extern Oid GetResGroupIdForRole(Oid roleid);
-extern void GetResGroupCapabilities(Oid groupId, ResGroupCaps *resgroupCaps);
+extern void GetResGroupCapabilities(Relation rel,
+									Oid groupId,
+									ResGroupCaps *resgroupCaps);
 extern void AtEOXact_ResGroup(bool isCommit);
 
 #endif   /* RESGROUPCMDS_H */
diff --git a/src/include/utils/resgroup.h b/src/include/utils/resgroup.h
index 84276b78955e4019595be9e308c17b9d87005393..fe547e486ee7cd0dc08a2ff7f6960de634db3073 100644
--- a/src/include/utils/resgroup.h
+++ b/src/include/utils/resgroup.h
@@ -27,11 +27,7 @@
 /*
  * Resource group capability.
  */
-typedef struct ResGroupCap
-{
-	int		value;
-	int		proposed;
-} ResGroupCap;
+typedef int32 ResGroupCap;
 
 /*
  * Resource group capabilities.
@@ -60,33 +56,6 @@ typedef struct ResGroupCaps
 	ResGroupCap		memSpillRatio;
 } ResGroupCaps;
 
-/*
- * Resource group setting options.
- *
- * These can represent the effective settings of a resource group,
- * or the new settings from ALTER RESOURCE GROUP syntax.
- *
- * The properties must be in the same order as ResGroupLimitType.
- *
- * This struct can also be converted to an array of int32 so the fields
- * can be accessed via index and iterated with loop.
- *
- *     ResGroupOpts opts;
- *     int32 *array = (int32 *) &opts;
- *     opts.concurrency = 1;
- *     array[RESGROUP_LIMIT_TYPE_CONCURRENCY] = 2;
- *     Assert(opts.concurrency == 2);
- */
-typedef struct ResGroupOpts
-{
-	int32			__unknown;			/* placeholder, do not use it */
-	int32			concurrency;
-	int32			cpuRateLimit;
-	int32			memLimit;
-	int32			memSharedQuota;
-	int32			memSpillRatio;
-} ResGroupOpts;
-
 /*
  * GUC variables.
  */
@@ -134,7 +103,7 @@ extern void ResGroupControlInit(void);
 /* Load resource group information from catalog */
 extern void	InitResGroups(void);
 
-extern void AllocResGroupEntry(Oid groupId, const ResGroupOpts *opts);
+extern void AllocResGroupEntry(Oid groupId, const ResGroupCaps *caps);
 
 extern void SerializeResGroupInfo(StringInfo str);
 extern void DeserializeResGroupInfo(struct ResGroupCaps *capsOut,
@@ -151,8 +120,6 @@ extern void SwitchResGroupOnSegment(const char *buf, int len);
 /* Retrieve statistic information of type from resource group */
 extern Datum ResGroupGetStat(Oid groupId, ResGroupStatType type);
 
-extern void ResGroupOptsToCaps(const ResGroupOpts *optsIn, ResGroupCaps *capsOut);
-extern void ResGroupCapsToOpts(const ResGroupCaps *capsIn, ResGroupOpts *optsOut);
 extern void ResGroupDumpMemoryInfo(void);
 
 /* Check the memory limit of resource group */
@@ -166,12 +133,6 @@ extern void ResGroupAlterOnCommit(Oid groupId,
 								  ResGroupLimitType limittype,
 								  const ResGroupCaps *caps);
 extern void ResGroupCheckForDrop(Oid groupId, char *name);
-extern void ResGroupDecideMemoryCaps(int groupId,
-									 ResGroupCaps *caps,
-									 const ResGroupOpts *opts);
-extern void ResGroupDecideConcurrencyCaps(Oid groupId,
-										  ResGroupCaps *caps,
-										  const ResGroupOpts *opts);
 
 extern int32 ResGroupGetVmemLimitChunks(void);
 extern int32 ResGroupGetVmemChunkSizeInBits(void);
diff --git a/src/test/isolation2/expected/resgroup/resgroup_alter_memory_spill_ratio.out b/src/test/isolation2/expected/resgroup/resgroup_alter_memory_spill_ratio.out
index 510ba93ad8fc50ed4bd4d97a18ccfe509d04725b..29d413094a5738b57a91c316f6599ea069912675 100644
--- a/src/test/isolation2/expected/resgroup/resgroup_alter_memory_spill_ratio.out
+++ b/src/test/isolation2/expected/resgroup/resgroup_alter_memory_spill_ratio.out
@@ -60,7 +60,7 @@ rg_spill_test|20                 |20                          |81
 
 -- negative: memory_spill_ratio is larger than RESGROUP_MAX_MEMORY_SPILL_RATIO
 ALTER RESOURCE GROUP rg_spill_test SET MEMORY_SPILL_RATIO 101;
-ERROR:  memory spill ratio cannot be greater than 100
+ERROR:  memory_spill_ratio range is [0, 100]
 SELECT * FROM rg_spill_status;
 groupname    |memory_shared_quota|proposed_memory_shared_quota|memory_spill_ratio|proposed_memory_spill_ratio
 -------------+-------------------+----------------------------+------------------+---------------------------
diff --git a/src/test/isolation2/expected/resgroup/resgroup_concurrency.out b/src/test/isolation2/expected/resgroup/resgroup_concurrency.out
index 6a1cb828ae3d997a79390973a756733e66575528..c5f8a5b02b738d131d5e43de631b52c4417ba4ad 100644
--- a/src/test/isolation2/expected/resgroup/resgroup_concurrency.out
+++ b/src/test/isolation2/expected/resgroup/resgroup_concurrency.out
@@ -167,7 +167,7 @@ ALTER
 SELECT concurrency,proposed_concurrency FROM gp_toolkit.gp_resgroup_config WHERE groupname='rg_concurrency_test';
 concurrency|proposed_concurrency
 -----------+--------------------
-3          |2                   
+2          |2                   
 (1 row)
 -- When one transaction is finished, queueing transaction won't be woken up. There're 2 running transactions and 1 queueing transaction.
 24:END;
diff --git a/src/test/isolation2/expected/resgroup/resgroup_syntax.out b/src/test/isolation2/expected/resgroup/resgroup_syntax.out
index 68e1d118870fbf172839e4d0d96c9531fb8b634f..d5a9a4b801f9f8766b34fe213b85dfc3ff9e7850 100644
--- a/src/test/isolation2/expected/resgroup/resgroup_syntax.out
+++ b/src/test/isolation2/expected/resgroup/resgroup_syntax.out
@@ -270,9 +270,9 @@ CREATE
 ALTER RESOURCE GROUP admin_group SET CONCURRENCY 0;
 ERROR:  admin_group must have at least one concurrency
 ALTER RESOURCE GROUP rg_test_group SET CONCURRENCY -1;
-ERROR:  concurrency limit cannot be less than 0
+ERROR:  concurrency range is [0, 'max_connections']
 ALTER RESOURCE GROUP rg_test_group SET CONCURRENCY 26;
-ERROR:  concurrency limit cannot be greater than 'max_connections'
+ERROR:  concurrency range is [0, 'max_connections']
 ALTER RESOURCE GROUP rg_test_group SET CONCURRENCY -0.5;
 ERROR:  syntax error at or near "0.5"
 LINE 1: ALTER RESOURCE GROUP rg_test_group SET CONCURRENCY -0.5;
@@ -310,9 +310,9 @@ ERROR:  syntax error at or near "0.1"
 LINE 1: ALTER RESOURCE GROUP rg_test_group SET CPU_RATE_LIMIT -0.1;
                                                                ^
 ALTER RESOURCE GROUP rg_test_group SET CPU_RATE_LIMIT -1;
-ERROR:  cpu rate limit cannot be less than 1
+ERROR:  cpu_rate_limit range is [1, 100]
 ALTER RESOURCE GROUP rg_test_group SET CPU_RATE_LIMIT 0;
-ERROR:  cpu rate limit cannot be less than 1
+ERROR:  cpu_rate_limit range is [1, 100]
 ALTER RESOURCE GROUP rg_test_group SET CPU_RATE_LIMIT 0.7;
 ERROR:  syntax error at or near "0.7"
 LINE 1: ALTER RESOURCE GROUP rg_test_group SET CPU_RATE_LIMIT 0.7;
diff --git a/src/test/isolation2/expected/resgroup/resgroup_unlimit_memory_spill_ratio.out b/src/test/isolation2/expected/resgroup/resgroup_unlimit_memory_spill_ratio.out
index 86cda29a513e30d01af4dadfc5e175417cd7c772..be374cf5fd7037d88a154084ad4363494d76619d 100644
--- a/src/test/isolation2/expected/resgroup/resgroup_unlimit_memory_spill_ratio.out
+++ b/src/test/isolation2/expected/resgroup/resgroup_unlimit_memory_spill_ratio.out
@@ -40,9 +40,9 @@ ALTER
 ALTER RESOURCE GROUP rg_spill_test SET MEMORY_SPILL_RATIO 100;
 ALTER
 ALTER RESOURCE GROUP rg_spill_test SET MEMORY_SPILL_RATIO -1;
-ERROR:  memory spill ratio cannot be less than 0
+ERROR:  memory_spill_ratio range is [0, 100]
 ALTER RESOURCE GROUP rg_spill_test SET MEMORY_SPILL_RATIO 101;
-ERROR:  memory spill ratio cannot be greater than 100
+ERROR:  memory_spill_ratio range is [0, 100]
 
 DROP RESOURCE GROUP rg_spill_test;
 DROP
diff --git a/src/test/isolation2/output/resgroup/resgroup_alter_memory.source b/src/test/isolation2/output/resgroup/resgroup_alter_memory.source
index cfe5b893655f945df4790e1a76fd54e0b1fdbf67..00f051b18f18419d7bd83ffeb147fdc127a3b2b7 100644
--- a/src/test/isolation2/output/resgroup/resgroup_alter_memory.source
+++ b/src/test/isolation2/output/resgroup/resgroup_alter_memory.source
@@ -94,7 +94,7 @@ ALTER
 SELECT * FROM rg_mem_status;
 groupname      |memory_limit|proposed_memory_limit|memory_shared_quota|proposed_memory_shared_quota
 ---------------+------------+---------------------+-------------------+----------------------------
-rg1_memory_test|60          |60                   |70                 |80                          
+rg1_memory_test|60          |60                   |80                 |80                          
 (1 row)
 
 1q: ... <quitting>
@@ -227,7 +227,7 @@ ALTER
 SELECT * FROM rg_mem_status;
 groupname      |memory_limit|proposed_memory_limit|memory_shared_quota|proposed_memory_shared_quota
 ---------------+------------+---------------------+-------------------+----------------------------
-rg1_memory_test|50          |60                   |60                 |60                          
+rg1_memory_test|60          |60                   |60                 |60                          
 (1 row)
 
 ALTER RESOURCE GROUP rg1_memory_test SET MEMORY_LIMIT 40;
@@ -253,7 +253,7 @@ ALTER
 SELECT * FROM rg_mem_status;
 groupname      |memory_limit|proposed_memory_limit|memory_shared_quota|proposed_memory_shared_quota
 ---------------+------------+---------------------+-------------------+----------------------------
-rg1_memory_test|40          |10                   |60                 |60                          
+rg1_memory_test|10          |10                   |60                 |60                          
 (1 row)
 
 --
@@ -291,7 +291,7 @@ ALTER
 SELECT * FROM rg_mem_status;
 groupname      |memory_limit|proposed_memory_limit|memory_shared_quota|proposed_memory_shared_quota
 ---------------+------------+---------------------+-------------------+----------------------------
-rg1_memory_test|40          |40                   |60                 |20                          
+rg1_memory_test|40          |40                   |20                 |20                          
 (1 row)
 
 ALTER RESOURCE GROUP rg1_memory_test SET MEMORY_LIMIT 30;
@@ -302,7 +302,7 @@ ALTER
 SELECT * FROM rg_mem_status;
 groupname      |memory_limit|proposed_memory_limit|memory_shared_quota|proposed_memory_shared_quota
 ---------------+------------+---------------------+-------------------+----------------------------
-rg1_memory_test|40          |30                   |60                 |20                          
+rg1_memory_test|30          |30                   |20                 |20                          
 (1 row)
 
 1q: ... <quitting>
@@ -345,7 +345,7 @@ ALTER
 SELECT * FROM rg_mem_status;
 groupname      |memory_limit|proposed_memory_limit|memory_shared_quota|proposed_memory_shared_quota
 ---------------+------------+---------------------+-------------------+----------------------------
-rg1_memory_test|30          |40                   |0                  |0                           
+rg1_memory_test|40          |40                   |0                  |0                           
 rg2_memory_test|20          |20                   |0                  |0                           
 (2 rows)
 
@@ -410,8 +410,8 @@ BEGIN
 SELECT * FROM rg_mem_status;
 groupname      |memory_limit|proposed_memory_limit|memory_shared_quota|proposed_memory_shared_quota
 ---------------+------------+---------------------+-------------------+----------------------------
-rg1_memory_test|30          |15                   |0                  |0                           
-rg2_memory_test|30          |40                   |0                  |0                           
+rg1_memory_test|15          |15                   |0                  |0                           
+rg2_memory_test|40          |40                   |0                  |0                           
 (2 rows)
 SELECT * FROM rg_activity_status;
 rsgname        |waiting_reason|current_query        
@@ -431,8 +431,8 @@ rg2_memory_test|              |<IDLE> in transaction
 SELECT * FROM rg_mem_status;
 groupname      |memory_limit|proposed_memory_limit|memory_shared_quota|proposed_memory_shared_quota
 ---------------+------------+---------------------+-------------------+----------------------------
-rg1_memory_test|30          |15                   |0                  |0                           
-rg2_memory_test|30          |40                   |0                  |0                           
+rg1_memory_test|15          |15                   |0                  |0                           
+rg2_memory_test|40          |40                   |0                  |0                           
 (2 rows)
 SELECT * FROM rg_activity_status;
 rsgname        |waiting_reason|current_query        
@@ -451,8 +451,8 @@ BEGIN
 SELECT * FROM rg_mem_status;
 groupname      |memory_limit|proposed_memory_limit|memory_shared_quota|proposed_memory_shared_quota
 ---------------+------------+---------------------+-------------------+----------------------------
-rg1_memory_test|30          |15                   |0                  |0                           
-rg2_memory_test|30          |40                   |0                  |0                           
+rg1_memory_test|15          |15                   |0                  |0                           
+rg2_memory_test|40          |40                   |0                  |0                           
 (2 rows)
 SELECT * FROM rg_activity_status;
 rsgname        |waiting_reason|current_query        
@@ -536,8 +536,8 @@ SET
 SELECT * FROM rg_mem_status;
 groupname      |memory_limit|proposed_memory_limit|memory_shared_quota|proposed_memory_shared_quota
 ---------------+------------+---------------------+-------------------+----------------------------
-rg1_memory_test|30          |15                   |0                  |0                           
-rg2_memory_test|30          |40                   |0                  |0                           
+rg1_memory_test|15          |15                   |0                  |0                           
+rg2_memory_test|40          |40                   |0                  |0                           
 (2 rows)
 SELECT * FROM rg_activity_status;
 rsgname        |waiting_reason|current_query        
@@ -557,8 +557,8 @@ END
 SELECT * FROM rg_mem_status;
 groupname      |memory_limit|proposed_memory_limit|memory_shared_quota|proposed_memory_shared_quota
 ---------------+------------+---------------------+-------------------+----------------------------
-rg1_memory_test|30          |15                   |0                  |0                           
-rg2_memory_test|30          |40                   |0                  |0                           
+rg1_memory_test|15          |15                   |0                  |0                           
+rg2_memory_test|40          |40                   |0                  |0                           
 (2 rows)
 SELECT * FROM rg_activity_status;
 rsgname        |waiting_reason|current_query        
@@ -581,8 +581,8 @@ BEGIN
 SELECT * FROM rg_mem_status;
 groupname      |memory_limit|proposed_memory_limit|memory_shared_quota|proposed_memory_shared_quota
 ---------------+------------+---------------------+-------------------+----------------------------
-rg1_memory_test|30          |15                   |0                  |0                           
-rg2_memory_test|30          |40                   |0                  |0                           
+rg1_memory_test|15          |15                   |0                  |0                           
+rg2_memory_test|40          |40                   |0                  |0                           
 (2 rows)
 SELECT * FROM rg_activity_status;
 rsgname        |waiting_reason|current_query        
@@ -666,8 +666,8 @@ SET
 SELECT * FROM rg_mem_status;
 groupname      |memory_limit|proposed_memory_limit|memory_shared_quota|proposed_memory_shared_quota
 ---------------+------------+---------------------+-------------------+----------------------------
-rg1_memory_test|30          |20                   |60                 |60                          
-rg2_memory_test|30          |40                   |0                  |0                           
+rg1_memory_test|20          |20                   |60                 |60                          
+rg2_memory_test|40          |40                   |0                  |0                           
 (2 rows)
 SELECT * FROM rg_activity_status;
 rsgname        |waiting_reason|current_query        
@@ -693,7 +693,7 @@ SELECT * FROM rg_mem_status;
 groupname      |memory_limit|proposed_memory_limit|memory_shared_quota|proposed_memory_shared_quota
 ---------------+------------+---------------------+-------------------+----------------------------
 rg1_memory_test|20          |20                   |30                 |30                          
-rg2_memory_test|30          |40                   |0                  |0                           
+rg2_memory_test|40          |40                   |0                  |0                           
 (2 rows)
 SELECT * FROM rg_activity_status;
 rsgname        |waiting_reason|current_query        
@@ -790,8 +790,8 @@ SET
 SELECT * FROM rg_mem_status;
 groupname      |memory_limit|proposed_memory_limit|memory_shared_quota|proposed_memory_shared_quota
 ---------------+------------+---------------------+-------------------+----------------------------
-rg1_memory_test|30          |20                   |90                 |90                          
-rg2_memory_test|30          |40                   |0                  |0                           
+rg1_memory_test|20          |20                   |90                 |90                          
+rg2_memory_test|40          |40                   |0                  |0                           
 (2 rows)
 SELECT * FROM rg_activity_status;
 rsgname        |waiting_reason|current_query        
@@ -810,8 +810,8 @@ ALTER
 SELECT * FROM rg_mem_status;
 groupname      |memory_limit|proposed_memory_limit|memory_shared_quota|proposed_memory_shared_quota
 ---------------+------------+---------------------+-------------------+----------------------------
-rg1_memory_test|30          |20                   |90                 |30                          
-rg2_memory_test|30          |40                   |0                  |0                           
+rg1_memory_test|20          |20                   |30                 |30                          
+rg2_memory_test|40          |40                   |0                  |0                           
 (2 rows)
 SELECT * FROM rg_activity_status;
 rsgname        |waiting_reason|current_query        
@@ -834,8 +834,8 @@ BEGIN
 SELECT * FROM rg_mem_status;
 groupname      |memory_limit|proposed_memory_limit|memory_shared_quota|proposed_memory_shared_quota
 ---------------+------------+---------------------+-------------------+----------------------------
-rg1_memory_test|30          |20                   |90                 |30                          
-rg2_memory_test|30          |40                   |0                  |0                           
+rg1_memory_test|20          |20                   |30                 |30                          
+rg2_memory_test|40          |40                   |0                  |0                           
 (2 rows)
 SELECT * FROM rg_activity_status;
 rsgname        |waiting_reason|current_query