/*
 * TypeRename
 *		This renames a type
 *
 * Note: any associated array type is *not* renamed; caller must make
 * another call to handle that case.  Currently this is only used for
 * renaming types associated with tables, for which there are no arrays.
 */
void
TypeRename(Oid typeOid, const char *newTypeName)
{
	Relation		pg_type_desc;
	HeapTuple		tuple;
	Form_pg_type	form;
	cqContext	   *pcqCtx;
	cqContext		cqc, cqc2;

	pg_type_desc = heap_open(TypeRelationId, RowExclusiveLock);

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

	tuple = caql_getfirst(
			pcqCtx,
			cql("SELECT * FROM pg_type "
				" WHERE oid = :1 "
				" FOR UPDATE ",
				ObjectIdGetDatum(typeOid)));

	if (!HeapTupleIsValid(tuple))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("type with OID \"%d\" does not exist", typeOid)));

	form = (Form_pg_type) GETSTRUCT(tuple);
	if (namestrcmp(&(form->typname), newTypeName))
	{
		if (caql_getcount(
					caql_addrel(cqclr(&cqc2), pg_type_desc),
					cql("SELECT COUNT(*) FROM pg_type "
						" WHERE typname = :1 "
						" AND typnamespace = :2 ",
						CStringGetDatum((char *) newTypeName),
						ObjectIdGetDatum(form->typnamespace))))
		{
			ereport(ERROR,
					(errcode(ERRCODE_DUPLICATE_OBJECT),
					 errmsg("type \"%s\" already exists", newTypeName)));
		}

		namestrcpy(&(form->typname), newTypeName);

		caql_update_current(pcqCtx, tuple);
		/* update the system catalog indexes (implicit) */
	}

	heap_freetuple(tuple);
	heap_close(pg_type_desc, RowExclusiveLock);
}

/*
 * Removes the policy of a table from the gp_distribution_policy table

 * Does nothing if the policy doesn't exist.
 */
void
GpPolicyRemove(Oid tbloid)
{
	Relation	gp_policy_rel;
	cqContext	cqc;

    /*
     * Open and lock the gp_distribution_policy catalog.
     */
	gp_policy_rel = heap_open(GpPolicyRelationId, RowExclusiveLock);

	/* Delete the policy entry from the catalog. */
	(void) caql_getcount(
		caql_addrel(cqclr(&cqc), gp_policy_rel),
		cql("DELETE FROM gp_distribution_policy "
			" WHERE localoid = :1 ",
			ObjectIdGetDatum(tbloid)));

	/*
     * Close the gp_distribution_policy relcache entry without unlocking.
     * We have updated the catalog: consequently the lock must be held until
     * end of transaction.
     */
    heap_close(gp_policy_rel, NoLock);
}                               /* GpPolicyRemove */

/*
 * Remove the master standby.
 *
 * gp_remove_master_standby()
 *
 * Returns:
 *  true upon success otherwise false
 */
Datum
gp_remove_master_standby(PG_FUNCTION_ARGS)
{
	int numDel;
	cqContext	cqc;

	mirroring_sanity_check(SUPERUSER | MASTER_ONLY | UTILITY_MODE,
						   "gp_remove_master_standby");

	if (!standby_exists())
		elog(ERROR, "no master standby defined");

	Relation rel = heap_open(GpSegmentConfigRelationId, AccessExclusiveLock);
	numDel= caql_getcount(caql_addrel(cqclr(&cqc), rel),
				cql("DELETE FROM gp_segment_configuration "
					" WHERE role = :1",
					CharGetDatum(SEGMENT_ROLE_STANDBY_CONFIG)));

	elog(LOG, "Remove standby, count : %d.", numDel);

	heap_close(rel, NoLock);

	update_gp_master_mirroring("Not Configured");

	PG_RETURN_BOOL(true);
}

/*
 * Activate a standby. To do this, we need to change
 *
 * 1. Check that we're actually the standby
 * 2. Remove standby from gp_segment_configuration.
 *
 * gp_activate_standby()
 *
 * Returns:
 *  true upon success, otherwise throws error.
 */
Datum
gp_activate_standby(PG_FUNCTION_ARGS)
{
	cqContext cqc;
	int numDel;
	mirroring_sanity_check(SUPERUSER | UTILITY_MODE | STANDBY_ONLY,
						   PG_FUNCNAME_MACRO);

	if (!AmIStandby())
		elog(ERROR, "%s must be run on the standby master",
			 PG_FUNCNAME_MACRO);

	/* remove standby from gp_segment_configuration */
	Relation rel = heap_open(GpSegmentConfigRelationId, AccessExclusiveLock);
	numDel= caql_getcount(caql_addrel(cqclr(&cqc), rel),
				cql("DELETE FROM gp_segment_configuration "
					" WHERE role = :1",
					CharGetDatum(SEGMENT_ROLE_STANDBY_CONFIG)));

	elog(LOG, "Remove standby while activating it, count : %d.", numDel);

	heap_close(rel, NoLock);

	/* done */
	PG_RETURN_BOOL(true);
}

/*
 * Removes the policy of a table from the gp_distribution_policy table
 *
 * Callers must check that there actually *is* a policy for the relation.
 */
void
GpPolicyRemove(Oid tbloid)
{
	Relation	gp_policy_rel;
	cqContext	cqc;

    /*
     * Open and lock the gp_distribution_policy catalog.
     */
	gp_policy_rel = heap_open(GpPolicyRelationId, RowExclusiveLock);

	/* Delete the policy entry from the catalog. */
	if (0 == caql_getcount(
				caql_addrel(cqclr(&cqc), gp_policy_rel),
				cql("DELETE FROM gp_distribution_policy "
					" WHERE localoid = :1 ",
					ObjectIdGetDatum(tbloid))))
	{
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("distribution policy for relation \"%d\" does not exist",
						tbloid)));
	}

	/*
     * Close the gp_distribution_policy relcache entry without unlocking.
     * We have updated the catalog: consequently the lock must be held until
     * end of transaction.
     */
    heap_close(gp_policy_rel, NoLock);
}                               /* GpPolicyRemove */

void update_segment_status_by_id(uint32_t id, char status)
{
	/* we use AccessExclusiveLock to prevent races */
	Relation rel = heap_open(GpSegmentConfigRelationId, AccessExclusiveLock);
	HeapTuple tuple;
	cqContext	cqc;
	cqContext  *pcqCtx;

	Assert(status == 'u' || status == 'd');

	pcqCtx = caql_beginscan(caql_addrel(cqclr(&cqc), rel),
						cql("SELECT * FROM gp_segment_configuration "
							" WHERE registration_order = :1 "
							" FOR UPDATE ",
							Int32GetDatum(id)));

	tuple = caql_getnext(pcqCtx);

	if (tuple != NULL) {
		if (((Form_gp_segment_configuration)GETSTRUCT(tuple))->status != status) {
			((Form_gp_segment_configuration)GETSTRUCT(tuple))->status = status;
			caql_update_current(pcqCtx, tuple);
		}
	} else {
		elog(LOG, "Can not find segment id: %d when update its status", id);
	}

	caql_endscan(pcqCtx);

	heap_close(rel, NoLock);
}

/*
 * GetFastSequences
 *
 * Get a list of consecutive sequence numbers. The starting sequence
 * number is the maximal value between 'lastsequence' + 1 and minSequence.
 * The length of the list is given.
 *
 * If there is not such an entry for objid in the table, create
 * one here.
 *
 * The existing entry for objid in the table is updated with a new
 * lastsequence value.
 */
int64 GetFastSequences(Oid objid, int64 objmod,
					   int64 minSequence, int64 numSequences)
{
	Relation gp_fastsequence_rel;
	TupleDesc tupleDesc;
	HeapTuple tuple;
	cqContext	 cqc;
	int64 firstSequence = minSequence;
	Datum lastSequenceDatum;
	int64 newLastSequence;

	gp_fastsequence_rel = heap_open(FastSequenceRelationId, RowExclusiveLock);
	tupleDesc = RelationGetDescr(gp_fastsequence_rel);
	
	tuple = caql_getfirst(
			caql_addrel(cqclr(&cqc), gp_fastsequence_rel),
			cql("SELECT * FROM gp_fastsequence "
				" WHERE objid = :1 "
				" AND objmod = :2 "
				" FOR UPDATE ",
				ObjectIdGetDatum(objid),
				Int64GetDatum(objmod)));

	if (!HeapTupleIsValid(tuple))
	{
		newLastSequence = firstSequence + numSequences - 1;
	}
	else
	{
		bool isNull;

		lastSequenceDatum = heap_getattr(tuple, Anum_gp_fastsequence_last_sequence,
										tupleDesc, &isNull);
		
		if (isNull)
			ereport(ERROR,
					(errcode(ERRCODE_UNDEFINED_OBJECT),
					 errmsg("got an invalid lastsequence number: NULL")));
		
		if (DatumGetInt64(lastSequenceDatum) + 1 > firstSequence)
			firstSequence = DatumGetInt64(lastSequenceDatum) + 1;
		newLastSequence = firstSequence + numSequences - 1;
	}
	
	update_fastsequence(gp_fastsequence_rel, tuple, tupleDesc,
						objid, objmod, newLastSequence);

	if (HeapTupleIsValid(tuple))
	{
		heap_freetuple(tuple);
	}
		
	/* Refer to the comment at the end of InsertFastSequenceEntry. */
	heap_close(gp_fastsequence_rel, RowExclusiveLock);

	return firstSequence;
}

/*
 * Add a master standby.
 *
 * gp_add_master_standby(hostname, address)
 *
 * Args:
 *  hostname - as above
 *  address - as above
 *
 * Returns:
 *  dbid of the new standby
 */
Datum
gp_add_master_standby(PG_FUNCTION_ARGS)
{
	CdbComponentDatabaseInfo *master = NULL;
	Relation	gprel;
	Datum values[Natts_gp_segment_configuration];
	bool nulls[Natts_gp_segment_configuration];
	HeapTuple tuple;
	cqContext	cqc;
	cqContext  *pcqCtx = NULL;

	if (PG_ARGISNULL(0))
		elog(ERROR, "host name cannot be NULL");
	if (PG_ARGISNULL(1))
		elog(ERROR, "address cannot be NULL");

	mirroring_sanity_check(MASTER_ONLY | UTILITY_MODE,
						   "gp_add_master_standby");

	if (standby_exists())
		elog(ERROR, "only a single master standby may be defined");

	/* master */
	master = registration_order_get_dbinfo(MASTER_ORDER_ID);

	/* Lock exclusively to avoid concurrent changes */
	gprel = heap_open(GpSegmentConfigRelationId, AccessExclusiveLock);

	pcqCtx = caql_beginscan(
				caql_addrel(cqclr(&cqc), gprel),
				cql("INSERT INTO gp_segment_configuration ", NULL));

	MemSet(nulls, false, sizeof(nulls));

	values[Anum_gp_segment_configuration_registration_order - 1] = Int32GetDatum(STANDBY_ORDER_ID);
	values[Anum_gp_segment_configuration_role - 1] = CharGetDatum(SEGMENT_ROLE_STANDBY_CONFIG);
	values[Anum_gp_segment_configuration_status - 1] = CharGetDatum('u');
	values[Anum_gp_segment_configuration_port - 1] = Int32GetDatum(master->port);
	values[Anum_gp_segment_configuration_hostname - 1] = PG_GETARG_DATUM(0);
	values[Anum_gp_segment_configuration_address - 1] = PG_GETARG_DATUM(1);
	nulls[Anum_gp_segment_configuration_description - 1] = true;

	tuple = caql_form_tuple(pcqCtx, values, nulls);

	/* insert a new tuple */
	caql_insert(pcqCtx, tuple); /* implicit update of index as well */

	caql_endscan(pcqCtx);

	if(master)
		pfree(master);

	heap_close(gprel, NoLock);

	PG_RETURN_INT16(1);
}

/*
 * This tests if non-equal predicates also use index scan.
 */
void
test__caql_switch5(void **state)
{
	const char		   *query = "SELECT * FROM pg_attribute "
								"WHERE attrelid = :1 and attnum > :2";
	struct caql_hash_cookie *hash_cookie;
	cqContext			context = {0}, *pCtx;
	Datum				keys[] = {ObjectIdGetDatum(ProcedureRelationId),
								  Int16GetDatum(0)};
	cq_list			   *pcql = CaQL(query, 2, keys);
	RelationData		dummyrel;
	SysScanDescData		dummydesc;

	dummyrel.rd_id = AttributeRelationId;

	hash_cookie = cq_lookup(query, strlen(query), pcql);

	/*
	 * Add explicit relation, and set indexOK = true and syscache = false.
	 */
	pCtx = caql_addrel(cqclr(&context), &dummyrel);
	pCtx = caql_indexOK(pCtx, true);
	pCtx = caql_syscache(pCtx, false);

	/* setup ScanKeyInit */
	expect__ScanKeyInit(NULL, false,
						Anum_pg_attribute_attrelid, true,
						BTEqualStrategyNumber, true,
						F_OIDEQ, true,
						NULL, false);

	/* setup ScanKeyInit */
	expect__ScanKeyInit(NULL, false,
						Anum_pg_attribute_attnum, true,
						BTGreaterStrategyNumber, true,
						F_INT2GT, true,
						NULL, false);

	/* setup systable_beginscan */
	expect__systable_beginscan(&dummyrel, true,
							   AttributeRelidNumIndexId, true,
							   true, true,
							   SnapshotNow, true,
							   2, true,
							   NULL, false,
							   &dummydesc);

	pCtx = caql_switch(hash_cookie, pCtx, pcql);

	assert_true(pCtx != NULL);
	assert_true(pCtx->cq_sysScan == &dummydesc);
	assert_true(pCtx->cq_heap_rel == &dummyrel);
	assert_false(pCtx->cq_usesyscache);
	assert_true(pCtx->cq_useidxOK);
}

/*
 * SetRelationRuleStatus
 *		Set the value of the relation's relhasrules field in pg_class;
 *		if the relation is becoming a view, also adjust its relkind.
 *
 * NOTE: caller must be holding an appropriate lock on the relation.
 *
 * NOTE: an important side-effect of this operation is that an SI invalidation
 * message is sent out to all backends --- including me --- causing relcache
 * entries to be flushed or updated with the new set of rules for the table.
 * This must happen even if we find that no change is needed in the pg_class
 * row.
 */
void
SetRelationRuleStatus(Oid relationId, bool relHasRules,
					  bool relIsBecomingView)
{
	Relation	relationRelation;
	HeapTuple	tuple;
	Form_pg_class classForm;
	cqContext	cqc;
	cqContext  *pcqCtx;

	/*
	 * Find the tuple to update in pg_class, using syscache for the lookup.
	 */
	relationRelation = heap_open(RelationRelationId, RowExclusiveLock);

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

	tuple = caql_getfirst(
			pcqCtx,
			cql("SELECT * FROM pg_class "
				" WHERE oid = :1 "
				" FOR UPDATE ",
				ObjectIdGetDatum(relationId)));

	if (!HeapTupleIsValid(tuple))
		elog(ERROR, "cache lookup failed for relation %u", relationId);
	classForm = (Form_pg_class) GETSTRUCT(tuple);

	if (classForm->relhasrules != relHasRules ||
		(relIsBecomingView && classForm->relkind != RELKIND_VIEW))
	{
		/* Do the update */
		classForm->relhasrules = relHasRules;
		if (relIsBecomingView)
		{
			classForm->relkind = RELKIND_VIEW;
			classForm->relstorage = RELSTORAGE_VIRTUAL;
		}

		caql_update_current(pcqCtx, tuple); /* implicit update of index  */
	}
	else
	{
		/* no need to change tuple, but force relcache rebuild anyway */
		CacheInvalidateRelcacheByTuple(tuple);
	}

	heap_freetuple(tuple);
	heap_close(relationRelation, RowExclusiveLock);
}

/*
 * IsErrorTable
 *
 * Returns true if relid is used as an error table, which has dependent object
 * that is an external table.  Though it's not great we didn't have a clear
 * definition of Error Table, it satisfies the current requirements.
 */
bool
IsErrorTable(Relation rel)
{
	cqContext	   *pcqCtx, *pcqCtxExt, ctxExt;
	HeapTuple		tup;
	Relation		extrel;
	bool			result = false;

	/* fast path to quick check */
	if (!RelationIsHeap(rel))
		return false;

	/*
	 * Otherwise, go deeper and play with pg_depend...
	 */
	pcqCtx = caql_beginscan(NULL,
							cql("SELECT * FROM pg_depend "
								" WHERE refclassid = :1 "
								" AND refobjid = :2 "
								" AND refobjsubid = :3 ",
								ObjectIdGetDatum(RelationRelationId),
								ObjectIdGetDatum(RelationGetRelid(rel)),
								Int32GetDatum(0)));

	extrel = relation_open(ExtTableRelationId, AccessShareLock);
	pcqCtxExt = caql_addrel(cqclr(&ctxExt), extrel);

	while (HeapTupleIsValid(tup = caql_getnext(pcqCtx)))
	{
		Form_pg_depend dep = (Form_pg_depend) GETSTRUCT(tup);
		Oid				fmterrtbl;

		fmterrtbl = caql_getoid(pcqCtxExt,
								cql("SELECT fmterrtbl FROM pg_exttable "
									" WHERE reloid = :1",
									ObjectIdGetDatum(dep->objid)));
		if (RelationGetRelid(rel) == fmterrtbl)
		{
			result = true;
			break;
		}
	}

	relation_close(extrel, AccessShareLock);

	caql_endscan(pcqCtx);

	return result;
}

/*
 * Get datum representations of the attoptions field in pg_attribute_encoding
 * for the given relation.
 */
Datum *
get_rel_attoptions(Oid relid, AttrNumber max_attno)
{
	Form_pg_attribute attform;
	HeapTuple		tuple;
	cqContext		cqc;
	cqContext	   *pcqCtx;
	Datum		   *dats;
	Relation 		pgae = heap_open(AttributeEncodingRelationId,
									 AccessShareLock);

	/* used for attbyval and len below */
	attform = pgae->rd_att->attrs[Anum_pg_attribute_encoding_attoptions - 1];

	dats = palloc0(max_attno * sizeof(Datum));

	pcqCtx = caql_beginscan(
			caql_addrel(cqclr(&cqc), pgae),
			cql("SELECT * FROM pg_attribute_encoding "
				" WHERE attrelid = :1 ",
				ObjectIdGetDatum(relid)));

	while (HeapTupleIsValid(tuple = caql_getnext(pcqCtx)))
	{
		Form_pg_attribute_encoding a = 
			(Form_pg_attribute_encoding)GETSTRUCT(tuple);
		int16 attnum = a->attnum;
		Datum attoptions;
		bool isnull;

		Insist(attnum > 0 && attnum <= max_attno);

		attoptions = heap_getattr(tuple, Anum_pg_attribute_encoding_attoptions,
								  RelationGetDescr(pgae), &isnull);
		Insist(!isnull);

		dats[attnum - 1] = datumCopy(attoptions,
									 attform->attbyval,
									 attform->attlen);
	}

	caql_endscan(pcqCtx);

	heap_close(pgae, AccessShareLock);

	return dats;

}

/* convert schema type from old dboid to newdboid */
bool convert_schema_type(char *schema, Oid oldDboid, Oid newDboid)
{
	Assert(NSPDBOID_CURRENT == oldDboid || NSPDBOID_CURRENT == newDboid);
	Assert(HcatalogDbOid == oldDboid || HcatalogDbOid == newDboid);
	Assert(newDboid != oldDboid);

	/* create tuples for pg_class table */
	HeapTuple nsptup = NULL;
	HeapTuple copytup = NULL;
	cqContext	cqc;

	Assert(Gp_role != GP_ROLE_EXECUTE);

	Relation rel = heap_open(NamespaceRelationId, RowExclusiveLock);

	cqContext *pcqCtx = caql_addrel(cqclr(&cqc), rel);
	nsptup =  caql_getfirst(
			pcqCtx,
			cql("SELECT * FROM pg_namespace "
				" WHERE nspname = :1 and nspdboid = :2",
				CStringGetDatum(schema), ObjectIdGetDatum(oldDboid)));


	if (!HeapTupleIsValid(nsptup))
	{
		heap_close(rel, NoLock);
		return false;
	}

	Datum values[Natts_pg_namespace];
	bool nulls[Natts_pg_namespace];
	bool replaces[Natts_pg_namespace];
	
	MemSet(values, 0, sizeof(values));
	MemSet(nulls, false, sizeof(nulls));
	MemSet(replaces, false, sizeof(replaces));
	
	replaces[Anum_pg_namespace_nspdboid - 1] = true;
	values[Anum_pg_namespace_nspdboid - 1] = newDboid;
	copytup = caql_modify_current(pcqCtx, values, nulls, replaces);

	caql_update_current(pcqCtx, copytup);
	heap_close(rel, NoLock);

	heap_freetuple(copytup);
	
	return true;
}

/*
 * Test whether given name is currently used as a constraint name
 * for the given object (relation or domain).
 *
 * This is used to decide whether to accept a user-specified constraint name.
 * It is deliberately not the same test as ChooseConstraintName uses to decide
 * whether an auto-generated name is OK: here, we will allow it unless there
 * is an identical constraint name in use *on the same object*.
 *
 * NB: Caller should hold exclusive lock on the given object, else
 * this test can be fooled by concurrent additions.
 */
bool
ConstraintNameIsUsed(ConstraintCategory conCat, Oid objId,
					 Oid objNamespace, const char *conname)
{
	bool		found;
	Relation	conDesc;
	HeapTuple	tup;
	cqContext  *pcqCtx;
	cqContext	cqc;

	conDesc = heap_open(ConstraintRelationId, AccessShareLock);

	found = false;

	pcqCtx = caql_beginscan(
			caql_addrel(cqclr(&cqc), conDesc),
			cql("SELECT * FROM pg_constraint "
				" WHERE conname = :1 "
				" AND connamespace = :2 ",
				CStringGetDatum((char *) conname),
				ObjectIdGetDatum(objNamespace)));

	while (HeapTupleIsValid(tup = caql_getnext(pcqCtx)))
	{
		Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(tup);

		if (conCat == CONSTRAINT_RELATION && con->conrelid == objId)
		{
			found = true;
			break;
		}
		else if (conCat == CONSTRAINT_DOMAIN && con->contypid == objId)
		{
			found = true;
			break;
		}
	}

	caql_endscan(pcqCtx);
	heap_close(conDesc, AccessShareLock);

	return found;
}

/* ----------------------------------------------------------------
 * caql_begin_CacheList()
 * Return a catclist
 *
 * In general, catquery will choose the syscache when the cql
 * statement contains an equality predicate on *all* of the syscache
 * primary key index columns, eg:
 *
 *   cql("SELECT * FROM pg_amop WHERE amopopr = :1 and amopclaid = :2 ")
 *
 * will use the AMOPOPID syscache with index
 * AccessMethodOperatorIndexId.  However, the cql statement for a
 * list-search requires an equality predicate on a subset of the
 * initial columns of the index, with *all* of the index columns
 * specified in an ORDER BY clause, eg:
 *
 *   cql("SELECT * FROM pg_amop WHERE amopopr = :1 "
 *       " ORDER BY amopopr, amopclaid ")
 *
 * will use a syscache list-search if this cql statement is an
 * argument to caql_begin_CacheList().  However, the syscache will
 * *not* be used for this statement if it is supplied for
 * caql_beginscan(), since SearchSysCache() can only return (at most)
 * a single tuple.
 *
 * NOTE: caql_begin_CacheList() will assert (Insist!) at runtime if
 * the cql statement does not map to a syscache lookup.
 * NOTE: it may be possible to "collapse" this API into the existing
 * beginscan/getnext/endscan.
 * ----------------------------------------------------------------
 */
struct catclist *
caql_begin_CacheList(cqContext *pCtx0,
					 cq_list *pcql)
{
	const char*				 caql_str = pcql->caqlStr;
	const char*				 filenam  = pcql->filename;
	int						 lineno	  = pcql->lineno;
	struct caql_hash_cookie	*pchn	  = cq_lookup(caql_str, strlen(caql_str), pcql);
	cqContext				*pCtx;
	cqContext				 cqc;

	if (NULL == pchn)
		elog(ERROR, "invalid caql string: %s\nfile: %s, line %d",
			 caql_str, filenam, lineno);

	Assert(!pchn->bInsert); /* INSERT not allowed */
	Assert(!pchn->bUpdate); /* UPDATE not allowed */
	Assert(!pchn->bDelete); /* DELETE not allowed */

	/* use the provided context, or provide a clean local ctx  */
	if (pCtx0)
		pCtx = pCtx0;
	else
		pCtx = cqclr(&cqc);

	/* NOTE: for case of cache list search, we must use syscache */
	pCtx->cq_bCacheList = true;

	pCtx = caql_switch(pchn, pCtx, pcql);
	/* NOTE: caql_switch frees the pcql */

	/* NOTE: must use the SysCache */
	Insist (pCtx->cq_usesyscache);

	caql_heapclose(pCtx);

	return SearchSysCacheKeyArrayList(pCtx->cq_cacheId,
									  pCtx->cq_NumKeys,
									  pCtx->cq_cacheKeys);
}

/*
 * Master function to remove a segment from all catalogs
 */
static void
remove_segment(int16 order)
{
	cqContext cqc;
	int numDel;

	if(order == MASTER_ORDER_ID || order == STANDBY_ORDER_ID)
		return;

	/* remove this segment from gp_segment_configuration */
	Relation rel = heap_open(GpSegmentConfigRelationId, AccessExclusiveLock);
	numDel= caql_getcount(caql_addrel(cqclr(&cqc), rel),
				cql("DELETE FROM gp_segment_configuration "
					" WHERE registration_order = :1",
					Int16GetDatum(order)));

	elog(LOG, "Remove segment successfully! Count : %d. Registration order : %d",
			  numDel,
			  order);

	heap_close(rel, NoLock);
}

void
ExtProtocolDeleteByOid(Oid	protOid)
{		
	Relation	rel;
	cqContext	cqc;

	/*
	 * Search pg_extprotocol.  We use a heapscan here even though there is an
	 * index on oid, on the theory that pg_extprotocol will usually have just a
	 * few entries and so an indexed lookup is a waste of effort.
	 */
	rel = heap_open(ExtprotocolRelationId, RowExclusiveLock);

	if (0 == caql_getcount(
				caql_addrel(cqclr(&cqc), rel),
				cql("DELETE FROM pg_extprotocol "
					" WHERE oid = :1 ",
					ObjectIdGetDatum(protOid))))
	{
		elog(ERROR, "protocol %u could not be found", protOid);
	}
	heap_close(rel, NoLock);
}

/*
 * Sets the policy of a table into the gp_distribution_policy table
 * from a GpPolicy structure.
 *
 */
void
GpPolicyReplace(Oid tbloid, const GpPolicy *policy)
{
	Relation	gp_policy_rel;
	HeapTuple	gp_policy_tuple = NULL;
	cqContext	cqc;
	cqContext  *pcqCtx;

	ArrayType  *attrnums;

	bool		nulls[2];
	Datum		values[2];
	bool		repl[2];

	Insist(policy->ptype == POLICYTYPE_PARTITIONED);

    /*
     * Open and lock the gp_distribution_policy catalog.
     */
	gp_policy_rel = heap_open(GpPolicyRelationId, RowExclusiveLock);

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

	/*
	 * Convert C arrays into Postgres arrays.
	 */
	if (policy->nattrs > 0)
	{
		int			i;
		Datum	   *akey;

		akey = (Datum *) palloc(policy->nattrs * sizeof(Datum));
		for (i = 0; i < policy->nattrs; i++)
			akey[i] = Int16GetDatum(policy->attrs[i]);
		attrnums = construct_array(akey, policy->nattrs,
								   INT2OID, 2, true, 's');
	}
	else
	{
		attrnums = NULL;
	}

	nulls[0] = false;
	nulls[1] = false;
	values[0] = ObjectIdGetDatum(tbloid);

	if (attrnums)
		values[1] = PointerGetDatum(attrnums);
	else
		nulls[1] = true;
		
	repl[0] = false;
	repl[1] = true;


	/*
	 * Select by value of the localoid field
	 */
	gp_policy_tuple = caql_getfirst(
			pcqCtx,
			cql("SELECT * FROM gp_distribution_policy "
				" WHERE localoid = :1 "
				" FOR UPDATE ",
				ObjectIdGetDatum(tbloid)));

	/*
	 * Read first (and only ) tuple
	 */

	if (HeapTupleIsValid(gp_policy_tuple))
	{
		
		HeapTuple newtuple = caql_modify_current(pcqCtx, values,
												 nulls, repl);
								
		caql_update_current(pcqCtx, newtuple); 
		/* and Update indexes (implicit) */

		heap_freetuple(newtuple);
	}
	else
	{
		gp_policy_tuple = caql_form_tuple(pcqCtx, values, nulls);
		caql_insert(pcqCtx, gp_policy_tuple);
		/* and Update indexes (implicit) */
	}
	
	/*
     * Close the gp_distribution_policy relcache entry without unlocking.
     * We have updated the catalog: consequently the lock must be held until
     * end of transaction.
     */
    heap_close(gp_policy_rel, NoLock);
}                               /* GpPolicyReplace */

/*
 * get dbinfo by role
 * There should be only one master in gp_segment_configuration table, one standby at most.
 */
CdbComponentDatabaseInfo *
role_get_dbinfo(char role)
{
	HeapTuple tuple;
	Relation rel;
	cqContext	cqc;
	bool bOnly;
	CdbComponentDatabaseInfo *i = NULL;

	Assert(role == SEGMENT_ROLE_PRIMARY ||
		   role == SEGMENT_ROLE_MASTER_CONFIG ||
		   role == SEGMENT_ROLE_STANDBY_CONFIG);
	/*
	 * Can only run on a master node, this restriction is due to the reliance
	 * on the gp_segment_configuration table.  This may be able to be relaxed
	 * by switching to a different method of checking.
	 */
	if (!AmActiveMaster() && !AmStandbyMaster())
		elog(ERROR, "role_get_dbinfo() executed on execution segment");

	rel = heap_open(GpSegmentConfigRelationId, AccessShareLock);

	tuple = caql_getfirst_only(
			caql_addrel(cqclr(&cqc), rel),
			&bOnly,
			cql("SELECT * FROM gp_segment_configuration "
				" WHERE role = :1 ",
				CharGetDatum(role)));

	if (HeapTupleIsValid(tuple))
	{
		Datum attr;
		bool isNull;

		i = palloc(sizeof(CdbComponentDatabaseInfo));

		/*
		 * role
		 */
		attr = heap_getattr(tuple, Anum_gp_segment_configuration_role,
							RelationGetDescr(rel), &isNull);
		Assert(!isNull);
		i->role = DatumGetChar(attr);

		/*
		 * status
		 */
		attr = heap_getattr(tuple, Anum_gp_segment_configuration_status,
							RelationGetDescr(rel), &isNull);
		Assert(!isNull);
		i->status = DatumGetChar(attr);

		/*
		 * hostname
		 */
		attr = heap_getattr(tuple, Anum_gp_segment_configuration_hostname,
							RelationGetDescr(rel), &isNull);
		Assert(!isNull);
		i->hostname = TextDatumGetCString(attr);

		/*
		 * address
		 */
		attr = heap_getattr(tuple, Anum_gp_segment_configuration_address,
							RelationGetDescr(rel), &isNull);
		Assert(!isNull);
		i->address = TextDatumGetCString(attr);

		/*
		 * port
		 */
		attr = heap_getattr(tuple, Anum_gp_segment_configuration_port,
							RelationGetDescr(rel), &isNull);
		Assert(!isNull);
		i->port = DatumGetInt32(attr);

	}
	else
	{
		elog(ERROR, "could not find configuration entry for role %c", role);
	}

	heap_close(rel, NoLock);

	return i;
}

CdbComponentDatabaseInfo *
dbid_get_dbinfo(int16 dbid)
{
	HeapTuple tuple;
	Relation rel;
	cqContext	cqc;
	bool bOnly;
	CdbComponentDatabaseInfo *i = NULL;

	/*
	 * Can only run on a master node, this restriction is due to the reliance
	 * on the gp_segment_configuration table.  This may be able to be relaxed
	 * by switching to a different method of checking.
	 */
	if (GpIdentity.segindex != MASTER_CONTENT_ID)
		elog(ERROR, "dbid_get_dbinfo() executed on execution segment");

	rel = heap_open(GpSegmentConfigRelationId, AccessShareLock);

	tuple = caql_getfirst_only(
			caql_addrel(cqclr(&cqc), rel),
			&bOnly,
			cql("SELECT * FROM gp_segment_configuration "
				" WHERE dbid = :1 ",
				Int16GetDatum(dbid)));

	if (HeapTupleIsValid(tuple))
	{
		Datum attr;
		bool isNull;

		i = palloc(sizeof(CdbComponentDatabaseInfo));

		/*
		 * dbid
		 */
		attr = heap_getattr(tuple, Anum_gp_segment_configuration_dbid,
							RelationGetDescr(rel), &isNull);
		Assert(!isNull);
		i->dbid = DatumGetInt16(attr);

		/*
		 * content
		 */
		attr = heap_getattr(tuple, Anum_gp_segment_configuration_content,
							RelationGetDescr(rel), &isNull);
		Assert(!isNull);
		i->segindex = DatumGetInt16(attr);

		/*
		 * role
		 */
		attr = heap_getattr(tuple, Anum_gp_segment_configuration_role,
							RelationGetDescr(rel), &isNull);
		Assert(!isNull);
		i->role = DatumGetChar(attr);

		/*
		 * preferred-role
		 */
		attr = heap_getattr(tuple,
							Anum_gp_segment_configuration_preferred_role,
							RelationGetDescr(rel), &isNull);
		Assert(!isNull);
		i->preferred_role = DatumGetChar(attr);

		/*
		 * mode
		 */
		attr = heap_getattr(tuple, Anum_gp_segment_configuration_mode,
							RelationGetDescr(rel), &isNull);
		Assert(!isNull);
		i->mode = DatumGetChar(attr);

		/*
		 * status
		 */
		attr = heap_getattr(tuple, Anum_gp_segment_configuration_status,
							RelationGetDescr(rel), &isNull);
		Assert(!isNull);
		i->status = DatumGetChar(attr);

		/*
		 * hostname
		 */
		attr = heap_getattr(tuple, Anum_gp_segment_configuration_hostname,
							RelationGetDescr(rel), &isNull);
		Assert(!isNull);
		i->hostname = TextDatumGetCString(attr);

		/*
		 * address
		 */
		attr = heap_getattr(tuple, Anum_gp_segment_configuration_address,
							RelationGetDescr(rel), &isNull);
		Assert(!isNull);
		i->address = TextDatumGetCString(attr);
		
		/*
		 * port
//.........这里部分代码省略.........