src/backend/resourcemanager/resqueuecommand.c - hawq - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */

 #include "envswitch.h"
 #include "dynrm.h"
 #include "resqueuecommand.h"
 #include "miscadmin.h"
 #include "communication/rmcomm_QD2RM.h"
 #include "utils/linkedlist.h"

 #include "catalog/pg_resqueue.h"
 #include "utils/resscheduler.h"
 #include "commands/defrem.h"

 /*******************************************************************************
  * This file contains all functions for creating, altering and dropping resource
  * queue through SQL DDL statement. All statement information is saved in the
  * argument stmt.
  ******************************************************************************/

 void validateDDLAttributeOptions(List *options);

 /*
  * CREATE RESOURCE QUEUE statement handler.
  */
 void createResourceQueue(CreateQueueStmt *stmt)
 {
 	int 		 res 			 = FUNC_RETURN_OK;
 	static char  errorbuf[1024];
 	Relation	 pg_resqueue_rel;
 	cqContext	 cqc;

 	/* Permission check - only superuser can create queues. */
 	if (!superuser())
 	{
 		ereport(ERROR,
 				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 				 errmsg("must be superuser to create resource queues")));
 	}
 	/*
      * MPP-7960: We cannot run CREATE RESOURCE QUEUE inside a user transaction
      * block because the shared memory structures are not cleaned up on abort,
      * resulting in "leaked", unreachable queues.
      */

     if (Gp_role == GP_ROLE_DISPATCH)
     {
         PreventTransactionChain((void *) stmt, "CREATE RESOURCE QUEUE");
     }

     /* Validate options. */
     validateDDLAttributeOptions(stmt->options);

 	/*
 	 * Check for an illegal name ('none' is used to signify no queue in ALTER
 	 * ROLE).
 	 */
 	if (strcmp(stmt->queue, "none") == 0)
 	{
 		ereport(ERROR,
 				(errcode(ERRCODE_RESERVED_NAME),
 				 errmsg("resource queue name %s is reserved",
 						stmt->queue),
 				 errOmitLocation(true)));
 	}
 	/*
 	 * Check the pg_resqueue relation to be certain the queue doesn't already
 	 * exist.
 	 */
 	pg_resqueue_rel = heap_open(ResQueueRelationId, RowExclusiveLock);

 	if (caql_getcount(
 			caql_addrel(cqclr(&cqc), pg_resqueue_rel),
 			cql("SELECT COUNT(*) FROM pg_resqueue WHERE rsqname = :1",
 				CStringGetDatum(stmt->queue))))
 	{
 		ereport(ERROR,
 				(errcode(ERRCODE_DUPLICATE_OBJECT),
 				 errmsg("resource queue \'%s\' already exists",
 						stmt->queue)));
 	}

 	heap_close(pg_resqueue_rel, NoLock);
     /*
      * Build the create resource queue request and send it to HAWQ RM process.
      * Basically, HAWQ RM runs all necessary logic to verify the statement and
      * apply the change. Therefore, QD only sends out the original information
      * and waits for the response.
      */
 	int resourceid = 0;
 	res = createNewResourceContext(&resourceid);
 	if ( res != FUNC_RETURN_OK )
 	{
 		Assert( res == COMM2RM_CLIENT_FULL_RESOURCECONTEXT );
 		ereport(ERROR,
 				(errcode(ERRCODE_INTERNAL_ERROR),
 						 errmsg("too many resource contexts were created")));
 	}

 	/* Here, using user oid is more convenient. */
 	res = registerConnectionInRMByOID(resourceid,
 									  GetUserId(),
 									  errorbuf,
 									  sizeof(errorbuf));
 	if ( res != FUNC_RETURN_OK )
 	{
 		releaseResourceContextWithErrorReport(resourceid);
 		ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("%s", errorbuf)));
 	}

     res = manipulateResourceQueue(resourceid,
     							  stmt->queue,
     							  MANIPULATE_RESQUEUE_CREATE,
     							  stmt->options,
     							  errorbuf,
     							  sizeof(errorbuf));

 	/* We always unregister connection. */
 	unregisterConnectionInRMWithErrorReport(resourceid);

 	/* We always release resource context. */
 	releaseResourceContextWithErrorReport(resourceid);

 	if ( res != FUNC_RETURN_OK )
 	{
 		ereport(ERROR,
 				(errcode(IS_TO_RM_RPC_ERROR(res) ?
 						 ERRCODE_INTERNAL_ERROR :
 						 ERRCODE_INVALID_OBJECT_DEFINITION),
 				 errmsg("%s", errorbuf)));
 	}
 	elog(LOG, "Complete applying CREATE RESOURCE QUEUE statement.");
 }

 /*******************************************************************************
  * DROP RESOURCE QUEUE statement handler.
  * stmt[in]		The parsed statement tree.
  ******************************************************************************/
 void dropResourceQueue(DropQueueStmt *stmt)
 {
 	int res 		= FUNC_RETURN_OK;
 	char errorbuf[1024];
 	Relation	 pg_resqueue_rel;
 	HeapTuple	 tuple;
 	cqContext	 cqc;
 	cqContext	*pcqCtx;
 	Oid			 queueid;

 	/* Permission check - only superuser can create queues. */
 	if (!superuser())
 	{
 		ereport(ERROR,
 				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 				 errmsg("must be superuser to drop resource queues")));
 	}

 	/* Cannot DROP default and root queue  */
 	if ( strcmp(stmt->queue, RESOURCE_QUEUE_DEFAULT_QUEUE_NAME) == 0 ||
 		 strcmp(stmt->queue, RESOURCE_QUEUE_ROOT_QUEUE_NAME) == 0 )
 	{
 		ereport(ERROR,
 				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 				 errmsg("cannot drop system resource queue \'%s\'",
 						stmt->queue)));
 	}

 	/*
 	 * Check the pg_resqueue relation to be certain the queue already
 	 * exists.
 	 */
 	pg_resqueue_rel = heap_open(ResQueueRelationId, RowExclusiveLock);

 	pcqCtx = caql_addrel(cqclr(&cqc), pg_resqueue_rel);

 	tuple = caql_getfirst(pcqCtx,
 						  cql("SELECT * FROM pg_resqueue WHERE rsqname = :1 FOR UPDATE",
 						  CStringGetDatum(stmt->queue)));

 	if (!HeapTupleIsValid(tuple))
 		ereport(ERROR,
 				(errcode(ERRCODE_UNDEFINED_OBJECT),
 				 errmsg("resource queue \'%s\' does not exist",
 						stmt->queue)));

 	/* Remember the Oid */
 	queueid = HeapTupleGetOid(tuple);

 	/*
 	 * Check to see if any roles are in this queue.
 	 */
 	if (caql_getcount(
 			NULL,
 			cql("SELECT COUNT(*) FROM pg_authid WHERE rolresqueue = :1",
 				ObjectIdGetDatum(queueid))))
 	{
 		ereport(ERROR,
 				(errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
 				 errmsg("resource queue \'%s\' is used by at least one role",
 						stmt->queue)));
 	}

 	heap_close(pg_resqueue_rel, NoLock);

 	/*
 	 * MPP-7960: We cannot run DROP RESOURCE QUEUE inside a user transaction
 	 * block because the shared memory structures are not cleaned up on abort,
 	 * resulting in "leaked", unreachable queues.
 	 */

 	if (Gp_role == GP_ROLE_DISPATCH)
 	{
 		PreventTransactionChain((void *) stmt, "DROP RESOURCE QUEUE");
 	}

 	/*
 	 * Build the drop resource queue request and send it to HAWQ RM process.
 	 * Basically, HAWQ RM runs all necessary logic to verify the statement and
 	 * apply the change. Therefore, QD only sends out the original information
 	 * and waits for the response.
 	 */
 	int resourceid = 0;
 	res = createNewResourceContext(&resourceid);
 	if ( res != FUNC_RETURN_OK ) {
 		Assert( res == COMM2RM_CLIENT_FULL_RESOURCECONTEXT );
 		ereport(ERROR,
 				(errcode(ERRCODE_INTERNAL_ERROR),
 						 errmsg("too many resource contexts were created")));
 	}

 	/* Here, using user oid is more convenient. */
 	res = registerConnectionInRMByOID(resourceid,
 									  GetUserId(),
 									  errorbuf,
 									  sizeof(errorbuf));
 	if ( res != FUNC_RETURN_OK ) {
 		releaseResourceContextWithErrorReport(resourceid);
 		ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("%s", errorbuf)));
 	}

 	res = manipulateResourceQueue(resourceid,
 								  stmt->queue,
 								  MANIPULATE_RESQUEUE_DROP,
 								  NULL,
 								  errorbuf,
 								  sizeof(errorbuf));

 	/* We always unregister connection. */
 	unregisterConnectionInRMWithErrorReport(resourceid);

 	/* We always release resource context. */
 	releaseResourceContextWithErrorReport(resourceid);

 	if ( res != FUNC_RETURN_OK )
 	{
 		ereport(ERROR,
 				(errcode(IS_TO_RM_RPC_ERROR(res) ?
 						 ERRCODE_INTERNAL_ERROR :
 						 ERRCODE_INVALID_OBJECT_DEFINITION),
 				 errmsg("%s", errorbuf)));
 	}

 	elog(LOG, "Completed applying DROP RESOURCE QUEUE statement.");
 }

 /*******************************************************************************
  * ALTER RESOURCE QUEUE statement handler.
  * stmt[in]		The parsed statement tree.
  ******************************************************************************/
 void alterResourceQueue(AlterQueueStmt *stmt)
 {
 	int 		res 		= FUNC_RETURN_OK;
 	static char errorbuf[1024];

 	Relation    pg_resqueue_rel;
 	cqContext	cqc;

 	/* Permission check - only superuser can create queues. */
 	if (!superuser())
 	{
 		ereport(ERROR,
 				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 				 errmsg("must be superuser to alter resource queues")));
 	}

 	/* Cannot DROP default and root queue  */
 	if ( strcmp(stmt->queue, RESOURCE_QUEUE_ROOT_QUEUE_NAME) == 0 )
 	{
 		ereport(ERROR,
 				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 				 errmsg("cannot alter system resource queue \'%s\'",
 						stmt->queue)));
 	}

 	/*
 	 * MPP-7960: We cannot run ALTER RESOURCE QUEUE inside a user transaction
 	 * block because the shared memory structures are not cleaned up on abort,
 	 * resulting in "leaked", unreachable queues.
 	 */

 	if (Gp_role == GP_ROLE_DISPATCH)
 	{
 		PreventTransactionChain((void *) stmt, "ALTER RESOURCE QUEUE");
 	}

     /* Validate options. */
     validateDDLAttributeOptions(stmt->options);

 	/*
 	 * Check if resource queue exists
 	 */
 	pg_resqueue_rel = heap_open(ResQueueRelationId, RowExclusiveLock);

 	if (caql_getcount(
 			caql_addrel(cqclr(&cqc), pg_resqueue_rel),
 			cql("SELECT COUNT(*) FROM pg_resqueue WHERE rsqname = :1",
 				CStringGetDatum(stmt->queue))) == 0)
 	{
 		ereport(ERROR,
 				(errcode(ERRCODE_DUPLICATE_OBJECT),
 				 errmsg("resource queue \'%s\' does not exist",
 						stmt->queue)));
 	}

 	heap_close(pg_resqueue_rel, NoLock);

 	/*
 	 * Build the alter resource queue request and send it to HAWQ RM process.
 	 * Basically, HAWQ RM runs all necessary logic to verify the statement and
 	 * apply the change. Therefore, QD only sends out the original information
 	 * and waits for the response.
 	 */
 	int resourceid = 0;
 	res = createNewResourceContext(&resourceid);
 	if (res != FUNC_RETURN_OK) {
 		Assert(res == COMM2RM_CLIENT_FULL_RESOURCECONTEXT);
 		ereport(ERROR,
 				(errcode(ERRCODE_INTERNAL_ERROR),
 						errmsg("too many existing resource context")));
 	}

 	/* Here, using user oid is more convenient. */
 	res = registerConnectionInRMByOID(resourceid,
 									  GetUserId(),
 									  errorbuf,
 									  sizeof(errorbuf));
 	if ( res != FUNC_RETURN_OK ) {
 		releaseResourceContextWithErrorReport(resourceid);
 		ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("%s", errorbuf)));
 	}

 	res = manipulateResourceQueue(resourceid,
 								  stmt->queue,
 								  MANIPULATE_RESQUEUE_ALTER,
 								  stmt->options,
 								  errorbuf,
 								  sizeof(errorbuf));
 	/* We always unregister connection. */
 	unregisterConnectionInRMWithErrorReport(resourceid);

 	/* We always release resource context. */
 	releaseResourceContextWithErrorReport(resourceid);

 	if ( res != FUNC_RETURN_OK )
 	{
 		ereport(ERROR,
 				(errcode(IS_TO_RM_RPC_ERROR(res) ?
 						 ERRCODE_INTERNAL_ERROR :
 						 ERRCODE_INVALID_OBJECT_DEFINITION),
 				 errmsg("%s", errorbuf)));
 	}

 	elog(LOG, "Completed applying ALTER RESOURCE QUEUE statement.");
 }

 #define VALID_DDL_DUP(index, defel, targref)				   \
 		if (strcmp((defel)->defname, RSQDDLAttrNames[(index)]) == 0)		   \
 		{																	   \
 			if ((targref) != NULL)											   \
 			{																   \
 				ereport(ERROR,												   \
 						(errcode(ERRCODE_SYNTAX_ERROR),						   \
 						 errmsg("redundant attribute %s",					   \
 								RSQDDLAttrNames[(index)])));				   \
 			}																   \
 			(targref) = (defel);											   \
 			continue;														   \
 		}

 void validateDDLAttributeOptions(List *options)
 {
 	DefElem		*dparent		  	  = NULL;
 	DefElem     *dactivelimit 	  	  = NULL;
 	DefElem     *dmemorylimit 	  	  = NULL;
 	DefElem     *dcorelimit 	  	  = NULL;
 	DefElem		*dvsegresquota	  	  = NULL;
 	DefElem		*dallocpolicy	  	  = NULL;
 	DefElem		*dresovercommit  	  = NULL;
 	DefElem		*dnvsegupperlimit 	  = NULL;
 	DefElem		*dnvseglowerlimit 	  = NULL;
 	DefElem		*dnvsegupperlimitpseg = NULL;
 	DefElem		*dnvseglowerlimitpseg = NULL;

 	Cost		 activelimit 	 = INVALID_RES_LIMIT_THRESHOLD;
 	ListCell    *option			 = NULL;

 	/* Extract options from the statement node tree, check duplicate options. */
 	foreach(option, options)
 	{
 		DefElem    *defel = (DefElem *) lfirst(option);
 		VALID_DDL_DUP(RSQ_DDL_ATTR_PARENT, 						defel, dparent)
 		VALID_DDL_DUP(RSQ_DDL_ATTR_ACTIVE_STATMENTS, 			defel, dactivelimit)
 		VALID_DDL_DUP(RSQ_DDL_ATTR_MEMORY_LIMIT_CLUSTER, 		defel, dmemorylimit)
 		VALID_DDL_DUP(RSQ_DDL_ATTR_CORE_LIMIT_CLUSTER, 			defel, dcorelimit)
 		VALID_DDL_DUP(RSQ_DDL_ATTR_VSEG_RESOURCE_QUOTA, 		defel, dvsegresquota)
 		VALID_DDL_DUP(RSQ_DDL_ATTR_ALLOCATION_POLICY, 			defel, dallocpolicy)
 		VALID_DDL_DUP(RSQ_DDL_ATTR_RESOURCE_OVERCOMMIT_FACTOR,	defel, dresovercommit)
 		VALID_DDL_DUP(RSQ_DDL_ATTR_NVSEG_UPPER_LIMIT, 			defel, dnvsegupperlimit)
 		VALID_DDL_DUP(RSQ_DDL_ATTR_NVSEG_LOWER_LIMIT, 			defel, dnvseglowerlimit)
 		VALID_DDL_DUP(RSQ_DDL_ATTR_NVSEG_UPPER_LIMIT_PERSEG, 	defel, dnvsegupperlimitpseg)
 		VALID_DDL_DUP(RSQ_DDL_ATTR_NVSEG_LOWER_LIMIT_PERSEG, 	defel, dnvseglowerlimitpseg)
 	}

 	/* Perform range checks on the various thresholds.*/
 	if (dactivelimit)
 	{
 		activelimit = (Cost) defGetInt64(dactivelimit);
 		if ( activelimit < MINIMUM_RESQUEUE_ACTIVESTATS_N )
 		{
 			ereport(ERROR,
 					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 					 errmsg("%s is less than %d",
 						   RSQDDLAttrNames[RSQ_DDL_ATTR_ACTIVE_STATMENTS],
 						   MINIMUM_RESQUEUE_ACTIVESTATS_N)));
 		}
 	}

 	/* Memory and core expression must be the same. */
 	if (dmemorylimit && dcorelimit)
 	{
 		bool  need_free_mem  = false;
 		bool  need_free_core = false;
 		char *memory_limit   = defGetString(dmemorylimit, &need_free_mem);
 		char *core_limit     = defGetString(dcorelimit,   &need_free_core);

 		if (memory_limit != NULL && core_limit != NULL)
 		{
 			if(strcmp(memory_limit, core_limit) != 0)
 			{
 				if(need_free_mem) { free(memory_limit); }
 				if(need_free_core) { free(core_limit); }
 				ereport(ERROR,
 						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 						errmsg("the values of %s and %s are not identical",
 								RSQDDLAttrNames[RSQ_DDL_ATTR_MEMORY_LIMIT_CLUSTER],
 								RSQDDLAttrNames[RSQ_DDL_ATTR_CORE_LIMIT_CLUSTER])));
 			}
 		}
 		else
 		{
 			if(need_free_mem) { free(memory_limit); }
 			if(need_free_core) { free(core_limit); }
 			ereport(ERROR,
 					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 					errmsg("invalid value of %s or %s",
 						   RSQDDLAttrNames[RSQ_DDL_ATTR_MEMORY_LIMIT_CLUSTER],
 						   RSQDDLAttrNames[RSQ_DDL_ATTR_CORE_LIMIT_CLUSTER])));
 		}
 	}

 	/*
 	 * NVSEG_UPPER_LIMIT/NVSEG_LOWER_LIMIT has 0 as default value that means the
 	 * setting is not effective, otherwise, it must be greater than 0.
 	 */
 	int64_t nvsegupperlimit = -1;
 	int64_t nvseglowerlimit = -1;
 	if (dnvsegupperlimit != NULL)
 	{
 		nvsegupperlimit = defGetInt64(dnvsegupperlimit);
 		if (nvsegupperlimit < MINIMUM_RESQUEUE_NVSEG_UPPER_LIMIT_N)
 		{
 			ereport(ERROR,
 					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 					 errmsg("%s is less than %s",
 							RSQDDLAttrNames[RSQ_DDL_ATTR_NVSEG_UPPER_LIMIT],
 							MINIMUM_RESQUEUE_NVSEG_UPPER_LIMIT)));
 		}
 	}

 	if (dnvseglowerlimit != NULL)
 	{
 		nvseglowerlimit = defGetInt64(dnvseglowerlimit);
 		if (nvseglowerlimit < MINIMUM_RESQUEUE_NVSEG_LOWER_LIMIT_N)
 		{
 			ereport(ERROR,
 					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 					 errmsg("%s is less than %s",
 							RSQDDLAttrNames[RSQ_DDL_ATTR_NVSEG_LOWER_LIMIT],
 							MINIMUM_RESQUEUE_NVSEG_LOWER_LIMIT)));
 		}
 	}

 	/*
 	 * NVSEG_UPPER_LIMIT_PERSEG/NVSEG_LOWER_LIMIT_PERSEG has 0 as default value
 	 * that means the setting is not effective, otherwise, it must be greater
 	 * than 0.
 	 */
 	double nvsegupperlimitpseg = -1.0;
 	double nvseglowerlimitpseg = -1.0;
 	if (dnvsegupperlimitpseg != NULL)
 	{
 		nvsegupperlimitpseg = defGetNumeric(dnvsegupperlimitpseg);
 		if (nvsegupperlimitpseg < MINIMUM_RESQUEUE_NVSEG_UPPER_PERSEG_LIMIT_N)
 		{
 			ereport(ERROR,
 					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 					 errmsg("%s is less than %s",
 							RSQDDLAttrNames[RSQ_DDL_ATTR_NVSEG_UPPER_LIMIT_PERSEG],
 							MINIMUM_RESQUEUE_NVSEG_UPPER_PERSEG_LIMIT)));
 		}
 	}

 	if (dnvseglowerlimitpseg != NULL)
 	{
 		nvseglowerlimitpseg = defGetNumeric(dnvseglowerlimitpseg);
 		if (nvseglowerlimitpseg < MINIMUM_RESQUEUE_NVSEG_LOWER_PERSEG_LIMIT_N)
 		{
 			ereport(ERROR,
 					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 					 errmsg("%s is less than %s",
 							RSQDDLAttrNames[RSQ_DDL_ATTR_NVSEG_LOWER_LIMIT_PERSEG],
 							MINIMUM_RESQUEUE_NVSEG_LOWER_PERSEG_LIMIT)));
 		}
 	}


 	/* The resource upper factor must be no less than 1. */
 	if( dresovercommit != NULL)
 	{
 		double resovercommit = defGetNumeric(dresovercommit);
 		if (resovercommit < MINIMUM_RESQUEUE_OVERCOMMIT_N)
 		{
 			ereport(ERROR,
 					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 					 errmsg("%s is less than %s",
 							RSQDDLAttrNames[RSQ_DDL_ATTR_RESOURCE_OVERCOMMIT_FACTOR],
 							MINIMUM_RESQUEUE_OVERCOMMIT)));
 		}
 	}
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/

	#include "envswitch.h"
	#include "dynrm.h"
	#include "resqueuecommand.h"
	#include "miscadmin.h"
	#include "communication/rmcomm_QD2RM.h"
	#include "utils/linkedlist.h"

	#include "catalog/pg_resqueue.h"
	#include "utils/resscheduler.h"
	#include "commands/defrem.h"

	/*******************************************************************************
	* This file contains all functions for creating, altering and dropping resource
	* queue through SQL DDL statement. All statement information is saved in the
	* argument stmt.
	******************************************************************************/

	void validateDDLAttributeOptions(List *options);

	/*
	* CREATE RESOURCE QUEUE statement handler.
	*/
	void createResourceQueue(CreateQueueStmt *stmt)
	{
	int res = FUNC_RETURN_OK;
	static char errorbuf[1024];
	Relation pg_resqueue_rel;
	cqContext cqc;

	/* Permission check - only superuser can create queues. */
	if (!superuser())
	{
	ereport(ERROR,
	(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
	errmsg("must be superuser to create resource queues")));
	}
	/*
	* MPP-7960: We cannot run CREATE RESOURCE QUEUE inside a user transaction
	* block because the shared memory structures are not cleaned up on abort,
	* resulting in "leaked", unreachable queues.
	*/

	if (Gp_role == GP_ROLE_DISPATCH)
	{
	PreventTransactionChain((void *) stmt, "CREATE RESOURCE QUEUE");
	}

	/* Validate options. */
	validateDDLAttributeOptions(stmt->options);

	/*
	* Check for an illegal name ('none' is used to signify no queue in ALTER
	* ROLE).
	*/
	if (strcmp(stmt->queue, "none") == 0)
	{
	ereport(ERROR,
	(errcode(ERRCODE_RESERVED_NAME),
	errmsg("resource queue name %s is reserved",
	stmt->queue),
	errOmitLocation(true)));
	}
	/*
	* Check the pg_resqueue relation to be certain the queue doesn't already
	* exist.
	*/
	pg_resqueue_rel = heap_open(ResQueueRelationId, RowExclusiveLock);

	if (caql_getcount(
	caql_addrel(cqclr(&cqc), pg_resqueue_rel),
	cql("SELECT COUNT(*) FROM pg_resqueue WHERE rsqname = :1",
	CStringGetDatum(stmt->queue))))
	{
	ereport(ERROR,
	(errcode(ERRCODE_DUPLICATE_OBJECT),
	errmsg("resource queue \'%s\' already exists",
	stmt->queue)));
	}

	heap_close(pg_resqueue_rel, NoLock);
	/*
	* Build the create resource queue request and send it to HAWQ RM process.
	* Basically, HAWQ RM runs all necessary logic to verify the statement and
	* apply the change. Therefore, QD only sends out the original information
	* and waits for the response.
	*/
	int resourceid = 0;
	res = createNewResourceContext(&resourceid);
	if ( res != FUNC_RETURN_OK )
	{
	Assert( res == COMM2RM_CLIENT_FULL_RESOURCECONTEXT );
	ereport(ERROR,
	(errcode(ERRCODE_INTERNAL_ERROR),
	errmsg("too many resource contexts were created")));
	}

	/* Here, using user oid is more convenient. */
	res = registerConnectionInRMByOID(resourceid,
	GetUserId(),
	errorbuf,
	sizeof(errorbuf));
	if ( res != FUNC_RETURN_OK )
	{
	releaseResourceContextWithErrorReport(resourceid);
	ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("%s", errorbuf)));
	}

	res = manipulateResourceQueue(resourceid,
	stmt->queue,
	MANIPULATE_RESQUEUE_CREATE,
	stmt->options,
	errorbuf,
	sizeof(errorbuf));

	/* We always unregister connection. */
	unregisterConnectionInRMWithErrorReport(resourceid);

	/* We always release resource context. */
	releaseResourceContextWithErrorReport(resourceid);

	if ( res != FUNC_RETURN_OK )
	{
	ereport(ERROR,
	(errcode(IS_TO_RM_RPC_ERROR(res) ?
	ERRCODE_INTERNAL_ERROR :
	ERRCODE_INVALID_OBJECT_DEFINITION),
	errmsg("%s", errorbuf)));
	}
	elog(LOG, "Complete applying CREATE RESOURCE QUEUE statement.");
	}

	/*******************************************************************************
	* DROP RESOURCE QUEUE statement handler.
	* stmt[in] The parsed statement tree.
	******************************************************************************/
	void dropResourceQueue(DropQueueStmt *stmt)
	{
	int res = FUNC_RETURN_OK;
	char errorbuf[1024];
	Relation pg_resqueue_rel;
	HeapTuple tuple;
	cqContext cqc;
	cqContext *pcqCtx;
	Oid queueid;

	/* Permission check - only superuser can create queues. */
	if (!superuser())
	{
	ereport(ERROR,
	(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
	errmsg("must be superuser to drop resource queues")));
	}

	/* Cannot DROP default and root queue */
	if ( strcmp(stmt->queue, RESOURCE_QUEUE_DEFAULT_QUEUE_NAME) == 0 \|\|
	strcmp(stmt->queue, RESOURCE_QUEUE_ROOT_QUEUE_NAME) == 0 )
	{
	ereport(ERROR,
	(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
	errmsg("cannot drop system resource queue \'%s\'",
	stmt->queue)));
	}

	/*
	* Check the pg_resqueue relation to be certain the queue already
	* exists.
	*/
	pg_resqueue_rel = heap_open(ResQueueRelationId, RowExclusiveLock);

	pcqCtx = caql_addrel(cqclr(&cqc), pg_resqueue_rel);

	tuple = caql_getfirst(pcqCtx,
	cql("SELECT * FROM pg_resqueue WHERE rsqname = :1 FOR UPDATE",
	CStringGetDatum(stmt->queue)));

	if (!HeapTupleIsValid(tuple))
	ereport(ERROR,
	(errcode(ERRCODE_UNDEFINED_OBJECT),
	errmsg("resource queue \'%s\' does not exist",
	stmt->queue)));

	/* Remember the Oid */
	queueid = HeapTupleGetOid(tuple);

	/*
	* Check to see if any roles are in this queue.
	*/
	if (caql_getcount(
	NULL,
	cql("SELECT COUNT(*) FROM pg_authid WHERE rolresqueue = :1",
	ObjectIdGetDatum(queueid))))
	{
	ereport(ERROR,
	(errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
	errmsg("resource queue \'%s\' is used by at least one role",
	stmt->queue)));
	}

	heap_close(pg_resqueue_rel, NoLock);

	/*
	* MPP-7960: We cannot run DROP RESOURCE QUEUE inside a user transaction
	* block because the shared memory structures are not cleaned up on abort,
	* resulting in "leaked", unreachable queues.
	*/

	if (Gp_role == GP_ROLE_DISPATCH)
	{
	PreventTransactionChain((void *) stmt, "DROP RESOURCE QUEUE");
	}

	/*
	* Build the drop resource queue request and send it to HAWQ RM process.
	* Basically, HAWQ RM runs all necessary logic to verify the statement and
	* apply the change. Therefore, QD only sends out the original information
	* and waits for the response.
	*/
	int resourceid = 0;
	res = createNewResourceContext(&resourceid);
	if ( res != FUNC_RETURN_OK ) {
	Assert( res == COMM2RM_CLIENT_FULL_RESOURCECONTEXT );
	ereport(ERROR,
	(errcode(ERRCODE_INTERNAL_ERROR),
	errmsg("too many resource contexts were created")));
	}

	/* Here, using user oid is more convenient. */
	res = registerConnectionInRMByOID(resourceid,
	GetUserId(),
	errorbuf,
	sizeof(errorbuf));
	if ( res != FUNC_RETURN_OK ) {
	releaseResourceContextWithErrorReport(resourceid);
	ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("%s", errorbuf)));
	}

	res = manipulateResourceQueue(resourceid,
	stmt->queue,
	MANIPULATE_RESQUEUE_DROP,
	NULL,
	errorbuf,
	sizeof(errorbuf));

	/* We always unregister connection. */
	unregisterConnectionInRMWithErrorReport(resourceid);

	/* We always release resource context. */
	releaseResourceContextWithErrorReport(resourceid);

	if ( res != FUNC_RETURN_OK )
	{
	ereport(ERROR,
	(errcode(IS_TO_RM_RPC_ERROR(res) ?
	ERRCODE_INTERNAL_ERROR :
	ERRCODE_INVALID_OBJECT_DEFINITION),
	errmsg("%s", errorbuf)));
	}

	elog(LOG, "Completed applying DROP RESOURCE QUEUE statement.");
	}

	/*******************************************************************************
	* ALTER RESOURCE QUEUE statement handler.
	* stmt[in] The parsed statement tree.
	******************************************************************************/
	void alterResourceQueue(AlterQueueStmt *stmt)
	{
	int res = FUNC_RETURN_OK;
	static char errorbuf[1024];

	Relation pg_resqueue_rel;
	cqContext cqc;

	/* Permission check - only superuser can create queues. */
	if (!superuser())
	{
	ereport(ERROR,
	(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
	errmsg("must be superuser to alter resource queues")));
	}

	/* Cannot DROP default and root queue */
	if ( strcmp(stmt->queue, RESOURCE_QUEUE_ROOT_QUEUE_NAME) == 0 )
	{
	ereport(ERROR,
	(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
	errmsg("cannot alter system resource queue \'%s\'",
	stmt->queue)));
	}

	/*
	* MPP-7960: We cannot run ALTER RESOURCE QUEUE inside a user transaction
	* block because the shared memory structures are not cleaned up on abort,
	* resulting in "leaked", unreachable queues.
	*/

	if (Gp_role == GP_ROLE_DISPATCH)
	{
	PreventTransactionChain((void *) stmt, "ALTER RESOURCE QUEUE");
	}

	/* Validate options. */
	validateDDLAttributeOptions(stmt->options);

	/*
	* Check if resource queue exists
	*/
	pg_resqueue_rel = heap_open(ResQueueRelationId, RowExclusiveLock);

	if (caql_getcount(
	caql_addrel(cqclr(&cqc), pg_resqueue_rel),
	cql("SELECT COUNT(*) FROM pg_resqueue WHERE rsqname = :1",
	CStringGetDatum(stmt->queue))) == 0)
	{
	ereport(ERROR,
	(errcode(ERRCODE_DUPLICATE_OBJECT),
	errmsg("resource queue \'%s\' does not exist",
	stmt->queue)));
	}

	heap_close(pg_resqueue_rel, NoLock);

	/*
	* Build the alter resource queue request and send it to HAWQ RM process.
	* Basically, HAWQ RM runs all necessary logic to verify the statement and
	* apply the change. Therefore, QD only sends out the original information
	* and waits for the response.
	*/
	int resourceid = 0;
	res = createNewResourceContext(&resourceid);
	if (res != FUNC_RETURN_OK) {
	Assert(res == COMM2RM_CLIENT_FULL_RESOURCECONTEXT);
	ereport(ERROR,
	(errcode(ERRCODE_INTERNAL_ERROR),
	errmsg("too many existing resource context")));
	}

	/* Here, using user oid is more convenient. */
	res = registerConnectionInRMByOID(resourceid,
	GetUserId(),
	errorbuf,
	sizeof(errorbuf));
	if ( res != FUNC_RETURN_OK ) {
	releaseResourceContextWithErrorReport(resourceid);
	ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("%s", errorbuf)));
	}

	res = manipulateResourceQueue(resourceid,
	stmt->queue,
	MANIPULATE_RESQUEUE_ALTER,
	stmt->options,
	errorbuf,
	sizeof(errorbuf));
	/* We always unregister connection. */
	unregisterConnectionInRMWithErrorReport(resourceid);

	/* We always release resource context. */
	releaseResourceContextWithErrorReport(resourceid);

	if ( res != FUNC_RETURN_OK )
	{
	ereport(ERROR,
	(errcode(IS_TO_RM_RPC_ERROR(res) ?
	ERRCODE_INTERNAL_ERROR :
	ERRCODE_INVALID_OBJECT_DEFINITION),
	errmsg("%s", errorbuf)));
	}

	elog(LOG, "Completed applying ALTER RESOURCE QUEUE statement.");
	}

	#define VALID_DDL_DUP(index, defel, targref) \
	if (strcmp((defel)->defname, RSQDDLAttrNames[(index)]) == 0) \
	{ \
	if ((targref) != NULL) \
	{ \
	ereport(ERROR, \
	(errcode(ERRCODE_SYNTAX_ERROR), \
	errmsg("redundant attribute %s", \
	RSQDDLAttrNames[(index)]))); \
	} \
	(targref) = (defel); \
	continue; \
	}

	void validateDDLAttributeOptions(List *options)
	{
	DefElem *dparent = NULL;
	DefElem *dactivelimit = NULL;
	DefElem *dmemorylimit = NULL;
	DefElem *dcorelimit = NULL;
	DefElem *dvsegresquota = NULL;
	DefElem *dallocpolicy = NULL;
	DefElem *dresovercommit = NULL;
	DefElem *dnvsegupperlimit = NULL;
	DefElem *dnvseglowerlimit = NULL;
	DefElem *dnvsegupperlimitpseg = NULL;
	DefElem *dnvseglowerlimitpseg = NULL;

	Cost activelimit = INVALID_RES_LIMIT_THRESHOLD;
	ListCell *option = NULL;

	/* Extract options from the statement node tree, check duplicate options. */
	foreach(option, options)
	{
	DefElem defel = (DefElem ) lfirst(option);
	VALID_DDL_DUP(RSQ_DDL_ATTR_PARENT, defel, dparent)
	VALID_DDL_DUP(RSQ_DDL_ATTR_ACTIVE_STATMENTS, defel, dactivelimit)
	VALID_DDL_DUP(RSQ_DDL_ATTR_MEMORY_LIMIT_CLUSTER, defel, dmemorylimit)
	VALID_DDL_DUP(RSQ_DDL_ATTR_CORE_LIMIT_CLUSTER, defel, dcorelimit)
	VALID_DDL_DUP(RSQ_DDL_ATTR_VSEG_RESOURCE_QUOTA, defel, dvsegresquota)
	VALID_DDL_DUP(RSQ_DDL_ATTR_ALLOCATION_POLICY, defel, dallocpolicy)
	VALID_DDL_DUP(RSQ_DDL_ATTR_RESOURCE_OVERCOMMIT_FACTOR, defel, dresovercommit)
	VALID_DDL_DUP(RSQ_DDL_ATTR_NVSEG_UPPER_LIMIT, defel, dnvsegupperlimit)
	VALID_DDL_DUP(RSQ_DDL_ATTR_NVSEG_LOWER_LIMIT, defel, dnvseglowerlimit)
	VALID_DDL_DUP(RSQ_DDL_ATTR_NVSEG_UPPER_LIMIT_PERSEG, defel, dnvsegupperlimitpseg)
	VALID_DDL_DUP(RSQ_DDL_ATTR_NVSEG_LOWER_LIMIT_PERSEG, defel, dnvseglowerlimitpseg)
	}

	/* Perform range checks on the various thresholds.*/
	if (dactivelimit)
	{
	activelimit = (Cost) defGetInt64(dactivelimit);
	if ( activelimit < MINIMUM_RESQUEUE_ACTIVESTATS_N )
	{
	ereport(ERROR,
	(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
	errmsg("%s is less than %d",
	RSQDDLAttrNames[RSQ_DDL_ATTR_ACTIVE_STATMENTS],
	MINIMUM_RESQUEUE_ACTIVESTATS_N)));
	}
	}

	/* Memory and core expression must be the same. */
	if (dmemorylimit && dcorelimit)
	{
	bool need_free_mem = false;
	bool need_free_core = false;
	char *memory_limit = defGetString(dmemorylimit, &need_free_mem);
	char *core_limit = defGetString(dcorelimit, &need_free_core);

	if (memory_limit != NULL && core_limit != NULL)
	{
	if(strcmp(memory_limit, core_limit) != 0)
	{
	if(need_free_mem) { free(memory_limit); }
	if(need_free_core) { free(core_limit); }
	ereport(ERROR,
	(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
	errmsg("the values of %s and %s are not identical",
	RSQDDLAttrNames[RSQ_DDL_ATTR_MEMORY_LIMIT_CLUSTER],
	RSQDDLAttrNames[RSQ_DDL_ATTR_CORE_LIMIT_CLUSTER])));
	}
	}
	else
	{
	if(need_free_mem) { free(memory_limit); }
	if(need_free_core) { free(core_limit); }
	ereport(ERROR,
	(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
	errmsg("invalid value of %s or %s",
	RSQDDLAttrNames[RSQ_DDL_ATTR_MEMORY_LIMIT_CLUSTER],
	RSQDDLAttrNames[RSQ_DDL_ATTR_CORE_LIMIT_CLUSTER])));
	}
	}

	/*
	* NVSEG_UPPER_LIMIT/NVSEG_LOWER_LIMIT has 0 as default value that means the
	* setting is not effective, otherwise, it must be greater than 0.
	*/
	int64_t nvsegupperlimit = -1;
	int64_t nvseglowerlimit = -1;
	if (dnvsegupperlimit != NULL)
	{
	nvsegupperlimit = defGetInt64(dnvsegupperlimit);
	if (nvsegupperlimit < MINIMUM_RESQUEUE_NVSEG_UPPER_LIMIT_N)
	{
	ereport(ERROR,
	(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
	errmsg("%s is less than %s",
	RSQDDLAttrNames[RSQ_DDL_ATTR_NVSEG_UPPER_LIMIT],
	MINIMUM_RESQUEUE_NVSEG_UPPER_LIMIT)));
	}
	}

	if (dnvseglowerlimit != NULL)
	{
	nvseglowerlimit = defGetInt64(dnvseglowerlimit);
	if (nvseglowerlimit < MINIMUM_RESQUEUE_NVSEG_LOWER_LIMIT_N)
	{
	ereport(ERROR,
	(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
	errmsg("%s is less than %s",
	RSQDDLAttrNames[RSQ_DDL_ATTR_NVSEG_LOWER_LIMIT],
	MINIMUM_RESQUEUE_NVSEG_LOWER_LIMIT)));
	}
	}

	/*
	* NVSEG_UPPER_LIMIT_PERSEG/NVSEG_LOWER_LIMIT_PERSEG has 0 as default value
	* that means the setting is not effective, otherwise, it must be greater
	* than 0.
	*/
	double nvsegupperlimitpseg = -1.0;
	double nvseglowerlimitpseg = -1.0;
	if (dnvsegupperlimitpseg != NULL)
	{
	nvsegupperlimitpseg = defGetNumeric(dnvsegupperlimitpseg);
	if (nvsegupperlimitpseg < MINIMUM_RESQUEUE_NVSEG_UPPER_PERSEG_LIMIT_N)
	{
	ereport(ERROR,
	(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
	errmsg("%s is less than %s",
	RSQDDLAttrNames[RSQ_DDL_ATTR_NVSEG_UPPER_LIMIT_PERSEG],
	MINIMUM_RESQUEUE_NVSEG_UPPER_PERSEG_LIMIT)));
	}
	}

	if (dnvseglowerlimitpseg != NULL)
	{
	nvseglowerlimitpseg = defGetNumeric(dnvseglowerlimitpseg);
	if (nvseglowerlimitpseg < MINIMUM_RESQUEUE_NVSEG_LOWER_PERSEG_LIMIT_N)
	{
	ereport(ERROR,
	(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
	errmsg("%s is less than %s",
	RSQDDLAttrNames[RSQ_DDL_ATTR_NVSEG_LOWER_LIMIT_PERSEG],
	MINIMUM_RESQUEUE_NVSEG_LOWER_PERSEG_LIMIT)));
	}
	}


	/* The resource upper factor must be no less than 1. */
	if( dresovercommit != NULL)
	{
	double resovercommit = defGetNumeric(dresovercommit);
	if (resovercommit < MINIMUM_RESQUEUE_OVERCOMMIT_N)
	{
	ereport(ERROR,
	(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
	errmsg("%s is less than %s",
	RSQDDLAttrNames[RSQ_DDL_ATTR_RESOURCE_OVERCOMMIT_FACTOR],
	MINIMUM_RESQUEUE_OVERCOMMIT)));
	}
	}
	}