/*
 * 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);
}

/*
 * Obtain the dbid of a of a segment at a given segment index (i.e., content id)
 * currently fulfilling the role specified. This means that the segment is
 * really performing the role of primary or mirror, irrespective of their
 * preferred role.
 */
int16
contentid_get_dbid(int16 contentid, char role, bool getPreferredRoleNotCurrentRole)
{
	int16 dbid = 0;
	bool bOnly;
	HeapTuple tup;

	/*
	 * 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, "contentid_get_dbid() executed on execution segment");

	/* XXX XXX: CHECK THIS  XXX jic 2011/12/09 */
	if (getPreferredRoleNotCurrentRole)
	{
		tup = caql_getfirst_only(
				NULL,
				&bOnly,
				cql("SELECT * FROM gp_segment_configuration "
					" WHERE content = :1 "
					" AND preferred_role = :2 ",
					Int16GetDatum(contentid),
					CharGetDatum(role)));
	}
	else
	{
		tup = caql_getfirst_only(
				NULL,
				&bOnly,
				cql("SELECT * FROM gp_segment_configuration "
					" WHERE content = :1 "
					" AND role = :2 ",
					Int16GetDatum(contentid),
					CharGetDatum(role)));
	}

	if (HeapTupleIsValid(tup))
	{
		dbid = ((Form_gp_segment_configuration) GETSTRUCT(tup))->dbid;
		/* We expect a single result, assert this */
		Assert(bOnly); /* should be only 1 */
	}
	/* no need to hold the lock, it's a catalog */

	return dbid;
}

/* ---------------------
 * addProcCallback() - Add a new callback to pg_proc_callback
 *
 * Parameters:
 *    profnoid    - oid of the function that has a callback
 *    procallback - oid of the callback function
 *    promethod   - role the callback function plays
 *
 * Notes:
 *    This function does not maintain dependencies in pg_depend, that behavior
 *    is currently controlled in pg_proc.c
 * ---------------------
 */
void 
addProcCallback(Oid profnoid, Oid procallback, char promethod)
{
	Relation	rel;
	bool		nulls[Natts_pg_proc_callback];
	Datum		values[Natts_pg_proc_callback];
	HeapTuple   tup;
	
	Insist(OidIsValid(profnoid));
	Insist(OidIsValid(procallback));

	/* open pg_proc_callback */
	rel = heap_open(ProcCallbackRelationId, RowExclusiveLock);

	/* Build the tuple and insert it */
	nulls[Anum_pg_proc_callback_profnoid - 1]	  = false;
	nulls[Anum_pg_proc_callback_procallback - 1]  = false;
	nulls[Anum_pg_proc_callback_promethod - 1]	  = false;
	values[Anum_pg_proc_callback_profnoid - 1]	  = ObjectIdGetDatum(profnoid);
	values[Anum_pg_proc_callback_procallback - 1] = ObjectIdGetDatum(procallback);
	values[Anum_pg_proc_callback_promethod - 1]	  = CharGetDatum(promethod);

	tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
	
	/* Insert tuple into the relation */
	simple_heap_insert(rel, tup);
	CatalogUpdateIndexes(rel, tup);

	heap_close(rel, RowExclusiveLock);
}

/* 
 * Returns the dbid of the mirror. We can use the fact that
 * mirrors have the same contentid (stored in GpIdentity) and go from
 * there.
 */
int16
my_mirror_dbid(void)
{
	int16 dbid = 0;
	int16 contentid = (int16)GpIdentity.segindex;
	bool bOnly;
	HeapTuple tup;

	/*
	 * 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, "my_mirror_dbid() executed on execution segment");

	tup = caql_getfirst_only(
			NULL,
			&bOnly,
			cql("SELECT dbid FROM gp_segment_configuration "
				" WHERE content = :1 "
				" AND role = :2 ",
				Int16GetDatum(contentid),
				CharGetDatum('m')));

	if (HeapTupleIsValid(tup))
	{
		dbid = ((Form_gp_segment_configuration) GETSTRUCT(tup))->dbid;
		/* We expect a single result, assert this */
		Assert(bOnly); /* should be only 1 */
	}
	/* no need to hold the lock, it's a catalog */

	return dbid;
}

/*
 * Gets list of all relation published by FOR ALL TABLES publication(s).
 */
List *
GetAllTablesPublicationRelations(void)
{
	Relation	classRel;
	ScanKeyData key[1];
	HeapScanDesc scan;
	HeapTuple	tuple;
	List	   *result = NIL;

	classRel = heap_open(RelationRelationId, AccessShareLock);

	ScanKeyInit(&key[0],
				Anum_pg_class_relkind,
				BTEqualStrategyNumber, F_CHAREQ,
				CharGetDatum(RELKIND_RELATION));

	scan = heap_beginscan_catalog(classRel, 1, key);

	while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
	{
		Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
		Oid			relid = relForm->oid;

		if (is_publishable_class(relid, relForm))
			result = lappend_oid(result, relid);
	}

	heap_endscan(scan);
	heap_close(classRel, AccessShareLock);

	return result;
}

/*
 * 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);
}

/*
 * Determine the dbid for the master standby
 */
int16
master_standby_dbid(void)
{
	int16 dbid = 0;
	int16 contentid = -1;
	bool bOnly;
	HeapTuple tup;

	/*
	 * Can only run on a master node, this restriction is due to the reliance
	 * on the gp_segment_configuration table.
	 */
	if (GpIdentity.segindex != MASTER_CONTENT_ID)
		elog(ERROR, "master_standby_dbid() executed on execution segment");

	tup = caql_getfirst_only(
			NULL,
			&bOnly,
			cql("SELECT * FROM gp_segment_configuration "
				" WHERE content = :1 "
				" AND role = :2 ",
				Int16GetDatum(contentid),
				CharGetDatum('m')));

	if (HeapTupleIsValid(tup))
	{
		dbid = ((Form_gp_segment_configuration) GETSTRUCT(tup))->dbid;
		/* We expect a single result, assert this */
		Assert(bOnly);
	}
	/* no need to hold the lock, it's a catalog */

	return dbid;
}

/*
 * InsertAgLabelTuple - register the new label in ag_label
 *
 * See InsertPgClassTuple()
 */
static void
InsertAgLabelTuple(Relation ag_label_desc, Oid laboid, RangeVar *label,
				   Oid relid, char labkind)
{
	Oid			graphid = get_graphname_oid(label->schemaname);
	char	   *labname = label->relname;
	int32		labid;
	Datum		values[Natts_ag_label];
	bool		nulls[Natts_ag_label];
	HeapTuple	tup;

	AssertArg(labkind == LABEL_KIND_VERTEX || labkind == LABEL_KIND_EDGE);

	labid = (int32) GetNewLabelId(label->schemaname, graphid);

	values[Anum_ag_label_labname - 1] = CStringGetDatum(labname);
	values[Anum_ag_label_graphid - 1] = CStringGetDatum(graphid);
	values[Anum_ag_label_labid - 1] = Int32GetDatum(labid);
	values[Anum_ag_label_relid - 1] = ObjectIdGetDatum(relid);
	values[Anum_ag_label_labkind - 1] = CharGetDatum(labkind);

	memset(nulls, false, sizeof(nulls));

	tup = heap_form_tuple(RelationGetDescr(ag_label_desc), values, nulls);

	HeapTupleSetOid(tup, laboid);

	simple_heap_insert(ag_label_desc, tup);

	CatalogUpdateIndexes(ag_label_desc, tup);

	heap_freetuple(tup);
}

/*
 * PgxcClassCreate
 *		Create a pgxc_class entry
 */
void
PgxcClassCreate(Oid pcrelid,
				char pclocatortype,
				int pcattnum,
				int pchashalgorithm,
				int pchashbuckets,
				int numnodes,
				Oid *nodes)
{
	Relation	pgxcclassrel;
	HeapTuple	htup;
	bool		nulls[Natts_pgxc_class];
	Datum		values[Natts_pgxc_class];
	int		i;
	oidvector	*nodes_array;

	/* Build array of Oids to be inserted */
	nodes_array = buildoidvector(nodes, numnodes);

	/* Iterate through attributes initializing nulls and values */
	for (i = 0; i < Natts_pgxc_class; i++)
	{
		nulls[i]  = false;
		values[i] = (Datum) 0;
	}

	/* should not happen */
	if (pcrelid == InvalidOid)
	{
		elog(ERROR,"pgxc class relid invalid.");
		return;
	}

	values[Anum_pgxc_class_pcrelid - 1]   = ObjectIdGetDatum(pcrelid);
	values[Anum_pgxc_class_pclocatortype - 1] = CharGetDatum(pclocatortype);

	if (pclocatortype == LOCATOR_TYPE_HASH || pclocatortype == LOCATOR_TYPE_MODULO)
	{
		values[Anum_pgxc_class_pcattnum - 1] = UInt16GetDatum(pcattnum);
		values[Anum_pgxc_class_pchashalgorithm - 1] = UInt16GetDatum(pchashalgorithm);
		values[Anum_pgxc_class_pchashbuckets - 1] = UInt16GetDatum(pchashbuckets);
	}

	/* Node information */
	values[Anum_pgxc_class_nodes - 1] = PointerGetDatum(nodes_array);

	/* Open the relation for insertion */
	pgxcclassrel = heap_open(PgxcClassRelationId, RowExclusiveLock);

	htup = heap_form_tuple(pgxcclassrel->rd_att, values, nulls);

	(void) simple_heap_insert(pgxcclassrel, htup);

	CatalogUpdateIndexes(pgxcclassrel, htup);

	heap_close(pgxcclassrel, RowExclusiveLock);
}

/*
 * Tell the caller whether a standby master is defined in the system.
 */
static bool
standby_exists()
{
	return caql_getcount(
			 NULL,
			 cql("SELECT COUNT(*) FROM gp_segment_configuration "
				 " WHERE role = :1 ",
				 CharGetDatum(SEGMENT_ROLE_STANDBY_CONFIG))) > 0;
}

/*
 * ConstraintIsAForeignKeyToReferenceTable function scans the pgConstraint to
 * fetch all of the constraints on the given relationId and see if at least one
 * of them is a foreign key referencing to a reference table.
 */
bool
ConstraintIsAForeignKeyToReferenceTable(char *constraintName, Oid relationId)
{
	Relation pgConstraint = NULL;
	SysScanDesc scanDescriptor = NULL;
	ScanKeyData scanKey[1];
	int scanKeyCount = 1;
	HeapTuple heapTuple = NULL;
	bool foreignKeyToReferenceTable = false;


	pgConstraint = heap_open(ConstraintRelationId, AccessShareLock);

	ScanKeyInit(&scanKey[0], Anum_pg_constraint_contype, BTEqualStrategyNumber, F_CHAREQ,
				CharGetDatum(CONSTRAINT_FOREIGN));
	scanDescriptor = systable_beginscan(pgConstraint, InvalidOid, false,
										NULL, scanKeyCount, scanKey);

	heapTuple = systable_getnext(scanDescriptor);
	while (HeapTupleIsValid(heapTuple))
	{
		Oid referencedTableId = InvalidOid;
		Form_pg_constraint constraintForm = (Form_pg_constraint) GETSTRUCT(heapTuple);
		char *constraintName = (constraintForm->conname).data;

		if (strncmp(constraintName, constraintName, NAMEDATALEN) != 0 ||
			constraintForm->conrelid != relationId)
		{
			heapTuple = systable_getnext(scanDescriptor);
			continue;
		}

		referencedTableId = constraintForm->confrelid;

		Assert(IsDistributedTable(referencedTableId));

		if (PartitionMethod(referencedTableId) == DISTRIBUTE_BY_NONE)
		{
			foreignKeyToReferenceTable = true;
			break;
		}

		heapTuple = systable_getnext(scanDescriptor);
	}

	/* clean up scan and close system catalog */
	systable_endscan(scanDescriptor);
	heap_close(pgConstraint, AccessShareLock);

	return foreignKeyToReferenceTable;
}

/*
 * Get all relations for subscription that are not in a ready state.
 *
 * Returned list is palloc'ed in current memory context.
 */
List *
GetSubscriptionNotReadyRelations(Oid subid)
{
	List	   *res = NIL;
	Relation	rel;
	HeapTuple	tup;
	int			nkeys = 0;
	ScanKeyData skey[2];
	SysScanDesc scan;

	rel = table_open(SubscriptionRelRelationId, AccessShareLock);

	ScanKeyInit(&skey[nkeys++],
				Anum_pg_subscription_rel_srsubid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(subid));

	ScanKeyInit(&skey[nkeys++],
				Anum_pg_subscription_rel_srsubstate,
				BTEqualStrategyNumber, F_CHARNE,
				CharGetDatum(SUBREL_STATE_READY));

	scan = systable_beginscan(rel, InvalidOid, false,
							  NULL, nkeys, skey);

	while (HeapTupleIsValid(tup = systable_getnext(scan)))
	{
		Form_pg_subscription_rel subrel;
		SubscriptionRelState *relstate;

		subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);

		relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
		relstate->relid = subrel->srrelid;
		relstate->state = subrel->srsubstate;
		relstate->lsn = subrel->srsublsn;

		res = lappend(res, relstate);
	}

	/* Cleanup */
	systable_endscan(scan);
	table_close(rel, AccessShareLock);

	return res;
}

/* ---------------------
 * lookupProcCallback() - Find a specified callback for a specified function
 *
 * Parameters:
 *    profnoid    - oid of the function that has a callback
 *    promethod   - which callback to find
 * ---------------------
 */
Oid  
lookupProcCallback(Oid profnoid, char promethod)
{
	Relation	rel;
	ScanKeyData skey[2];
	SysScanDesc scan;
	HeapTuple	tup;
	Oid         result;

	Insist(OidIsValid(profnoid));

	/* open pg_proc_callback */
	rel = heap_open(ProcCallbackRelationId, AccessShareLock);

	/* Lookup (profnoid, promethod) from index */
	/* (profnoid, promethod) is guaranteed unique by the index */
	ScanKeyInit(&skey[0],
				Anum_pg_proc_callback_profnoid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(profnoid));
	ScanKeyInit(&skey[1],
				Anum_pg_proc_callback_promethod,
				BTEqualStrategyNumber, F_CHAREQ,
				CharGetDatum(promethod));
	scan = systable_beginscan(rel, ProcCallbackProfnoidPromethodIndexId, true,
							  SnapshotNow, 2, skey);
	tup = systable_getnext(scan);
	if (HeapTupleIsValid(tup))
	{
		Datum		d;
		bool		isnull;

		d = heap_getattr(tup, Anum_pg_proc_callback_procallback,
						 RelationGetDescr(rel), &isnull);
		Assert(!isnull);

		result = DatumGetObjectId(d);
	}
	else
		result = InvalidOid;

	systable_endscan(scan);
	heap_close(rel, AccessShareLock);

	return result;
}

/*
 * Update the state of a subscription table.
 */
void
UpdateSubscriptionRelState(Oid subid, Oid relid, char state,
						   XLogRecPtr sublsn)
{
	Relation	rel;
	HeapTuple	tup;
	bool		nulls[Natts_pg_subscription_rel];
	Datum		values[Natts_pg_subscription_rel];
	bool		replaces[Natts_pg_subscription_rel];

	LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);

	rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);

	/* Try finding existing mapping. */
	tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
							  ObjectIdGetDatum(relid),
							  ObjectIdGetDatum(subid));
	if (!HeapTupleIsValid(tup))
		elog(ERROR, "subscription table %u in subscription %u does not exist",
			 relid, subid);

	/* Update the tuple. */
	memset(values, 0, sizeof(values));
	memset(nulls, false, sizeof(nulls));
	memset(replaces, false, sizeof(replaces));

	replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
	values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);

	replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
	if (sublsn != InvalidXLogRecPtr)
		values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
	else
		nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;

	tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
							replaces);

	/* Update the catalog. */
	CatalogTupleUpdate(rel, &tup->t_self, tup);

	/* Cleanup. */
	table_close(rel, NoLock);
}

/*
 * Add new state record for a subscription table.
 */
void
AddSubscriptionRelState(Oid subid, Oid relid, char state,
						XLogRecPtr sublsn)
{
	Relation	rel;
	HeapTuple	tup;
	bool		nulls[Natts_pg_subscription_rel];
	Datum		values[Natts_pg_subscription_rel];

	LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);

	rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);

	/* Try finding existing mapping. */
	tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
							  ObjectIdGetDatum(relid),
							  ObjectIdGetDatum(subid));
	if (HeapTupleIsValid(tup))
		elog(ERROR, "subscription table %u in subscription %u already exists",
			 relid, subid);

	/* Form the tuple. */
	memset(values, 0, sizeof(values));
	memset(nulls, false, sizeof(nulls));
	values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(subid);
	values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(relid);
	values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
	if (sublsn != InvalidXLogRecPtr)
		values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
	else
		nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;

	tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);

	/* Insert tuple into catalog. */
	CatalogTupleInsert(rel, tup);

	heap_freetuple(tup);

	/* Cleanup. */
	table_close(rel, NoLock);
}

/*
 * ConstraintIsAForeignKey returns true if the given constraint name
 * is a foreign key to defined on the relation.
 */
bool
ConstraintIsAForeignKey(char *constraintNameInput, Oid relationId)
{
	Relation pgConstraint = NULL;
	SysScanDesc scanDescriptor = NULL;
	ScanKeyData scanKey[1];
	int scanKeyCount = 1;
	HeapTuple heapTuple = NULL;

	pgConstraint = heap_open(ConstraintRelationId, AccessShareLock);

	ScanKeyInit(&scanKey[0], Anum_pg_constraint_contype, BTEqualStrategyNumber, F_CHAREQ,
				CharGetDatum(CONSTRAINT_FOREIGN));
	scanDescriptor = systable_beginscan(pgConstraint, InvalidOid, false,
										NULL, scanKeyCount, scanKey);

	heapTuple = systable_getnext(scanDescriptor);
	while (HeapTupleIsValid(heapTuple))
	{
		Form_pg_constraint constraintForm = (Form_pg_constraint) GETSTRUCT(heapTuple);
		char *constraintName = (constraintForm->conname).data;

		if (strncmp(constraintName, constraintNameInput, NAMEDATALEN) == 0 &&
			constraintForm->conrelid == relationId)
		{
			systable_endscan(scanDescriptor);
			heap_close(pgConstraint, AccessShareLock);

			return true;
		}

		heapTuple = systable_getnext(scanDescriptor);
	}

	/* clean up scan and close system catalog */
	systable_endscan(scanDescriptor);
	heap_close(pgConstraint, AccessShareLock);

	return false;
}

/*
 * Returns pre-defined hidden tuples for pg_proc.
 */
HeapTuple *
GetHiddenPgProcTuples(Relation pg_proc, int *len)
{
	HeapTuple	   *tuples;
	Datum			values[Natts_pg_proc];
	bool			nulls[Natts_pg_proc];
	MemoryContext	oldcontext;
	static HeapTuple *StaticPgProcTuples = NULL;
	static int StaticPgProcTupleLen = 0;

	if (StaticPgProcTuples != NULL)
	{
		*len = StaticPgProcTupleLen;
		return StaticPgProcTuples;
	}

#define N_PGPROC_TUPLES 2
	oldcontext = MemoryContextSwitchTo(CacheMemoryContext);
	tuples = palloc(sizeof(HeapTuple) * N_PGPROC_TUPLES);

	/*
	 * gp_read_error_log
	 *
	 * CREATE FUNCTION pg_catalog.gp_read_error_log(
	 *     text,
	 *     cmdtime OUT timestamptz,
	 *     relname OUT text,
	 *     filename OUT text,
	 *     linenum OUT int4,
	 *     bytenum OUT int4,
	 *     errmsg OUT text,
	 *     rawdata OUT text,
	 *     rawbytes OUT bytea
	 *   ) RETURNS SETOF record AS 'gp_read_error_log'
	 *   LANGUAGE internal VOLATILE STRICT <RUN ON SEGMENT>;
	 */
	{
		NameData		procname = {"gp_read_error_log"};
		Oid				proargtypes[] = {TEXTOID};
		ArrayType	   *array;
		Datum			allargtypes[9];
		Datum			proargmodes[9];
		Datum			proargnames[9];

		MemSet(nulls, false, sizeof(bool) * Natts_pg_proc);
		values[Anum_pg_proc_proname - 1] = NameGetDatum(&procname);
		values[Anum_pg_proc_pronamespace - 1] = ObjectIdGetDatum(PG_CATALOG_NAMESPACE);
		values[Anum_pg_proc_proowner - 1] = ObjectIdGetDatum(BOOTSTRAP_SUPERUSERID);
		values[Anum_pg_proc_prolang - 1] = ObjectIdGetDatum(INTERNALlanguageId);
		values[Anum_pg_proc_proisagg - 1] = BoolGetDatum(false);
		values[Anum_pg_proc_prosecdef - 1] = BoolGetDatum(false);
		values[Anum_pg_proc_proisstrict - 1] = BoolGetDatum(true);
		values[Anum_pg_proc_proretset - 1] = BoolGetDatum(true);
		values[Anum_pg_proc_provolatile - 1] = CharGetDatum('v');
		values[Anum_pg_proc_pronargs - 1] = Int16GetDatum(1);
		values[Anum_pg_proc_prorettype - 1] = ObjectIdGetDatum(RECORDOID);
		values[Anum_pg_proc_proiswin - 1] = BoolGetDatum(false);
		values[Anum_pg_proc_proargtypes - 1] = PointerGetDatum(buildoidvector(proargtypes, 1));

		allargtypes[0] = TEXTOID;
		allargtypes[1] = TIMESTAMPTZOID;
		allargtypes[2] = TEXTOID;
		allargtypes[3] = TEXTOID;
		allargtypes[4] = INT4OID;
		allargtypes[5] = INT4OID;
		allargtypes[6] = TEXTOID;
		allargtypes[7] = TEXTOID;
		allargtypes[8] = BYTEAOID;
		array = construct_array(allargtypes, 9, OIDOID, 4, true, 'i');
		values[Anum_pg_proc_proallargtypes - 1] = PointerGetDatum(array);

		proargmodes[0] = CharGetDatum('i');
		proargmodes[1] = CharGetDatum('o');
		proargmodes[2] = CharGetDatum('o');
		proargmodes[3] = CharGetDatum('o');
		proargmodes[4] = CharGetDatum('o');
		proargmodes[5] = CharGetDatum('o');
		proargmodes[6] = CharGetDatum('o');
		proargmodes[7] = CharGetDatum('o');
		proargmodes[8] = CharGetDatum('o');
		array = construct_array(proargmodes, 9, CHAROID, 1, true, 'c');
		values[Anum_pg_proc_proargmodes - 1] = PointerGetDatum(array);

		proargnames[0] = CStringGetTextDatum("");
		proargnames[1] = CStringGetTextDatum("cmdtime");
		proargnames[2] = CStringGetTextDatum("relname");
		proargnames[3] = CStringGetTextDatum("filename");
		proargnames[4] = CStringGetTextDatum("linenum");
		proargnames[5] = CStringGetTextDatum("bytenum");
		proargnames[6] = CStringGetTextDatum("errmsg");
		proargnames[7] = CStringGetTextDatum("rawdata");
		proargnames[8] = CStringGetTextDatum("rawbytes");
		array = construct_array(proargnames, 9, TEXTOID, -1, false, 'i');
		values[Anum_pg_proc_proargnames - 1] = PointerGetDatum(array);
		values[Anum_pg_proc_prosrc - 1] = CStringGetTextDatum("gp_read_error_log");
		values[Anum_pg_proc_probin - 1] = (Datum) 0;
		nulls[Anum_pg_proc_probin - 1] = true;
//.........这里部分代码省略.........

//.........这里部分代码省略.........
							 "aggregate_dummy", /* placeholder proc */
							 NULL,		/* probin */
							 true,		/* isAgg */
							 false,		/* isWindowFunc */
							 false,		/* security invoker (currently not
										 * definable for agg) */
							 false,		/* isLeakProof */
							 false,		/* isStrict (not needed for agg) */
							 PROVOLATILE_IMMUTABLE,		/* volatility (not
														 * needed for agg) */
							 proparallel,
							 parameterTypes,	/* paramTypes */
							 allParameterTypes, /* allParamTypes */
							 parameterModes,	/* parameterModes */
							 parameterNames,	/* parameterNames */
							 parameterDefaults, /* parameterDefaults */
							 PointerGetDatum(NULL),		/* trftypes */
							 PointerGetDatum(NULL),		/* proconfig */
							 1, /* procost */
							 0);	/* prorows */
	procOid = myself.objectId;

	/*
	 * Okay to create the pg_aggregate entry.
	 */

	/* initialize nulls and values */
	for (i = 0; i < Natts_pg_aggregate; i++)
	{
		nulls[i] = false;
		values[i] = (Datum) NULL;
	}
	values[Anum_pg_aggregate_aggfnoid - 1] = ObjectIdGetDatum(procOid);
	values[Anum_pg_aggregate_aggkind - 1] = CharGetDatum(aggKind);
	values[Anum_pg_aggregate_aggnumdirectargs - 1] = Int16GetDatum(numDirectArgs);
	values[Anum_pg_aggregate_aggtransfn - 1] = ObjectIdGetDatum(transfn);
	values[Anum_pg_aggregate_aggfinalfn - 1] = ObjectIdGetDatum(finalfn);
	values[Anum_pg_aggregate_aggcombinefn - 1] = ObjectIdGetDatum(combinefn);
	values[Anum_pg_aggregate_aggserialfn - 1] = ObjectIdGetDatum(serialfn);
	values[Anum_pg_aggregate_aggdeserialfn - 1] = ObjectIdGetDatum(deserialfn);
	values[Anum_pg_aggregate_aggmtransfn - 1] = ObjectIdGetDatum(mtransfn);
	values[Anum_pg_aggregate_aggminvtransfn - 1] = ObjectIdGetDatum(minvtransfn);
	values[Anum_pg_aggregate_aggmfinalfn - 1] = ObjectIdGetDatum(mfinalfn);
	values[Anum_pg_aggregate_aggfinalextra - 1] = BoolGetDatum(finalfnExtraArgs);
	values[Anum_pg_aggregate_aggmfinalextra - 1] = BoolGetDatum(mfinalfnExtraArgs);
	values[Anum_pg_aggregate_aggsortop - 1] = ObjectIdGetDatum(sortop);
	values[Anum_pg_aggregate_aggtranstype - 1] = ObjectIdGetDatum(aggTransType);
	values[Anum_pg_aggregate_aggserialtype - 1] = ObjectIdGetDatum(aggSerialType);
	values[Anum_pg_aggregate_aggtransspace - 1] = Int32GetDatum(aggTransSpace);
	values[Anum_pg_aggregate_aggmtranstype - 1] = ObjectIdGetDatum(aggmTransType);
	values[Anum_pg_aggregate_aggmtransspace - 1] = Int32GetDatum(aggmTransSpace);
	if (agginitval)
		values[Anum_pg_aggregate_agginitval - 1] = CStringGetTextDatum(agginitval);
	else
		nulls[Anum_pg_aggregate_agginitval - 1] = true;
	if (aggminitval)
		values[Anum_pg_aggregate_aggminitval - 1] = CStringGetTextDatum(aggminitval);
	else
		nulls[Anum_pg_aggregate_aggminitval - 1] = true;

	aggdesc = heap_open(AggregateRelationId, RowExclusiveLock);
	tupDesc = aggdesc->rd_att;

	tup = heap_form_tuple(tupDesc, values, nulls);
	simple_heap_insert(aggdesc, tup);

/*
 * InsertRule -
 *	  takes the arguments and inserts them as a row into the system
 *	  relation "pg_rewrite"
 */
static Oid
InsertRule(char *rulname,
		   int evtype,
		   Oid eventrel_oid,
		   AttrNumber evslot_index,
		   bool evinstead,
		   Node *event_qual,
		   List *action,
		   bool replace)
{
	char	   *evqual = nodeToString(event_qual);
	char	   *actiontree = nodeToString((Node *) action);
	int			i;
	Datum		values[Natts_pg_rewrite];
	bool		nulls[Natts_pg_rewrite];
	bool		replaces[Natts_pg_rewrite];
	NameData	rname;
	Relation	pg_rewrite_desc;
	HeapTuple	tup,
				oldtup;
	Oid			rewriteObjectId;
	ObjectAddress myself,
				referenced;
	bool		is_update = false;

	/*
	 * Set up *nulls and *values arrays
	 */
	MemSet(nulls, false, sizeof(nulls));

	i = 0;
	namestrcpy(&rname, rulname);
	values[i++] = NameGetDatum(&rname); /* rulename */
	values[i++] = ObjectIdGetDatum(eventrel_oid);		/* ev_class */
	values[i++] = Int16GetDatum(evslot_index);	/* ev_attr */
	values[i++] = CharGetDatum(evtype + '0');	/* ev_type */
	values[i++] = CharGetDatum(RULE_FIRES_ON_ORIGIN);	/* ev_enabled */
	values[i++] = BoolGetDatum(evinstead);		/* is_instead */
	values[i++] = CStringGetTextDatum(evqual);	/* ev_qual */
	values[i++] = CStringGetTextDatum(actiontree);		/* ev_action */

	/*
	 * Ready to store new pg_rewrite tuple
	 */
	pg_rewrite_desc = heap_open(RewriteRelationId, RowExclusiveLock);

	/*
	 * Check to see if we are replacing an existing tuple
	 */
	oldtup = SearchSysCache2(RULERELNAME,
							 ObjectIdGetDatum(eventrel_oid),
							 PointerGetDatum(rulname));

	if (HeapTupleIsValid(oldtup))
	{
		if (!replace)
			ereport(ERROR,
					(errcode(ERRCODE_DUPLICATE_OBJECT),
					 errmsg("rule \"%s\" for relation \"%s\" already exists",
							rulname, get_rel_name(eventrel_oid))));

		/*
		 * When replacing, we don't need to replace every attribute
		 */
		MemSet(replaces, false, sizeof(replaces));
		replaces[Anum_pg_rewrite_ev_attr - 1] = true;
		replaces[Anum_pg_rewrite_ev_type - 1] = true;
		replaces[Anum_pg_rewrite_is_instead - 1] = true;
		replaces[Anum_pg_rewrite_ev_qual - 1] = true;
		replaces[Anum_pg_rewrite_ev_action - 1] = true;

		tup = heap_modify_tuple(oldtup, RelationGetDescr(pg_rewrite_desc),
								values, nulls, replaces);

		simple_heap_update(pg_rewrite_desc, &tup->t_self, tup);

		ReleaseSysCache(oldtup);

		rewriteObjectId = HeapTupleGetOid(tup);
		is_update = true;
	}
	else
	{
		tup = heap_form_tuple(pg_rewrite_desc->rd_att, values, nulls);

		rewriteObjectId = simple_heap_insert(pg_rewrite_desc, tup);
	}

	/* Need to update indexes in either case */
	CatalogUpdateIndexes(pg_rewrite_desc, tup);

	heap_freetuple(tup);

	/* If replacing, get rid of old dependencies and make new ones */
	if (is_update)
//.........这里部分代码省略.........