static void
PersistentBuild_FindGpRelationNodeIndex(
	Oid				database,

	Oid				defaultTablespace,

	RelFileNode		*relFileNode)
{
	Relation	pg_class_rel;

	HeapScanDesc scan;
	HeapTuple	tuple;

	bool found;

	/*
	 * Iterate through all the relations of the database and find gp_relation_node_index.
	 */
	pg_class_rel = 
			DirectOpen_PgClassOpen(
							defaultTablespace, 
							database);
	scan = heap_beginscan(pg_class_rel, SnapshotNow, 0, NULL);
	found = false;
	while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
	{
		Oid 			relationOid;

		Form_pg_class	form_pg_class;

		Oid 			reltablespace;

		relationOid = HeapTupleGetOid(tuple);
		if (relationOid != GpRelfileNodeOidIndexId)
		{
			continue;
		}

		form_pg_class = (Form_pg_class) GETSTRUCT(tuple);

		reltablespace = form_pg_class->reltablespace;

		if (reltablespace == 0)
		{
			reltablespace = defaultTablespace;
		}

		relFileNode->spcNode = reltablespace;
		relFileNode->dbNode = database;
		relFileNode->relNode= form_pg_class->relfilenode;

		found = true;
		break;
	}
	heap_endscan(scan);

	DirectOpen_PgClassClose(pg_class_rel);

	if (!found)
	{
		elog(ERROR, "pg_class entry for gp_relation_node_index not found");
	}

}

Type Type_fromOid(Oid typeId, jobject typeMap)
{
	CacheEntry   ce;
	HeapTuple    typeTup;
	Form_pg_type typeStruct;
	Type         type = Type_fromOidCache(typeId);

	if(type != 0)
		return type;

	typeTup    = PgObject_getValidTuple(TYPEOID, typeId, "type");
	typeStruct = (Form_pg_type)GETSTRUCT(typeTup);

	if(typeStruct->typelem != 0 && typeStruct->typlen == -1)
	{
		type = Type_getArrayType(Type_fromOid(typeStruct->typelem, typeMap), typeId);
		goto finally;
	}

	/* For some reason, the anyarray is *not* an array with anyelement as the
	 * element type. We'd like to see it that way though.
	 */
	if(typeId == ANYARRAYOID)
	{
		type = Type_getArrayType(Type_fromOid(ANYELEMENTOID, typeMap), typeId);
		goto finally;
	}

	if(typeStruct->typbasetype != 0)
	{
		/* Domain type, recurse using the base type (which in turn may
		 * also be a domain)
		 */
		type = Type_fromOid(typeStruct->typbasetype, typeMap);
		goto finally;
	}

	if(typeMap != 0)
	{
		jobject joid      = Oid_create(typeId);
		jclass  typeClass = (jclass)JNI_callObjectMethod(typeMap, s_Map_get, joid);

		JNI_deleteLocalRef(joid);
		if(typeClass != 0)
		{
			TupleDesc tupleDesc = lookup_rowtype_tupdesc_noerror(typeId, -1, true);
			type = (Type)UDT_registerUDT(typeClass, typeId, typeStruct, tupleDesc, false);
			JNI_deleteLocalRef(typeClass);
			goto finally;
		}
	}

	/* Composite and record types will not have a TypeObtainer registered
	 */
	if(typeStruct->typtype == 'c' || (typeStruct->typtype == 'p' && typeId == RECORDOID))
	{
		type = Composite_obtain(typeId);
		goto finally;
	}

	ce = (CacheEntry)HashMap_getByOid(s_obtainerByOid, typeId);
	if(ce == 0)
		/*
		 * Default to String and standard textin/textout coersion.
		 */
		type = String_obtain(typeId);
	else
	{
		type = ce->type;
		if(type == 0)
			type = ce->obtainer(typeId);
	}

finally:
	ReleaseSysCache(typeTup);
	Type_cacheByOid(typeId, type);
	return type;
}

/*
 * Finds an external protocol by passed in protocol name.
 * Errors if no such protocol exist, or if no function to
 * execute this protocol exists (for read or write separately).
 * 
 * Returns the protocol function to use.
 */
Oid
LookupExtProtocolFunction(const char *prot_name, 
						  ExtPtcFuncType prot_type,
						  bool error)
{
	Relation	rel;
	Oid			funcOid = InvalidOid;
	ScanKeyData skey;
	SysScanDesc scan;
	HeapTuple	tup;
	Form_pg_extprotocol extprot;

	rel = heap_open(ExtprotocolRelationId, AccessShareLock);

	/*
	 * Check the pg_extprotocol relation to be certain the protocol
	 * is there.
	 */
	ScanKeyInit(&skey,
				Anum_pg_extprotocol_ptcname,
				BTEqualStrategyNumber, F_NAMEEQ,
				CStringGetDatum(prot_name));
	scan = systable_beginscan(rel, ExtprotocolPtcnameIndexId, true,
							  SnapshotNow, 1, &skey);
	tup = systable_getnext(scan);

	if (!HeapTupleIsValid(tup))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("protocol \"%s\" does not exist",
						prot_name)));

	extprot = (Form_pg_extprotocol) GETSTRUCT(tup);

	switch (prot_type)
	{
		case EXTPTC_FUNC_READER:
			funcOid = extprot->ptcreadfn;
			break;
		case EXTPTC_FUNC_WRITER:
			funcOid = extprot->ptcwritefn;
			break;
		case EXTPTC_FUNC_VALIDATOR:
			funcOid = extprot->ptcvalidatorfn;
			break;
		default:
			elog(ERROR, "internal error in pg_extprotocol:func_type_to_attnum");
			break;
	}
	systable_endscan(scan);
	heap_close(rel, AccessShareLock);

	if (!OidIsValid(funcOid) && error)
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("protocol '%s' has no %s function defined",
						 prot_name, func_type_to_name(prot_type))));

	return funcOid;

}

/*
 * sepgsql_relation_post_create
 *
 * The post creation hook of relation/attribute
 */
void
sepgsql_relation_post_create(Oid relOid)
{
	Relation	rel;
	ScanKeyData skey;
	SysScanDesc sscan;
	HeapTuple	tuple;
	Form_pg_class classForm;
	ObjectAddress object;
	uint16		tclass;
	const char *tclass_text;
	char	   *scontext;		/* subject */
	char	   *tcontext;		/* schema */
	char	   *rcontext;		/* relation */
	char	   *ccontext;		/* column */
	char		audit_name[2 * NAMEDATALEN + 20];

	/*
	 * Fetch catalog record of the new relation. Because pg_class entry is not
	 * visible right now, we need to scan the catalog using SnapshotSelf.
	 */
	rel = heap_open(RelationRelationId, AccessShareLock);

	ScanKeyInit(&skey,
				ObjectIdAttributeNumber,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(relOid));

	sscan = systable_beginscan(rel, ClassOidIndexId, true,
							   SnapshotSelf, 1, &skey);

	tuple = systable_getnext(sscan);
	if (!HeapTupleIsValid(tuple))
		elog(ERROR, "catalog lookup failed for relation %u", relOid);

	classForm = (Form_pg_class) GETSTRUCT(tuple);

	switch (classForm->relkind)
	{
		case RELKIND_RELATION:
			tclass = SEPG_CLASS_DB_TABLE;
			tclass_text = "table";
			break;
		case RELKIND_SEQUENCE:
			tclass = SEPG_CLASS_DB_SEQUENCE;
			tclass_text = "sequence";
			break;
		case RELKIND_VIEW:
			tclass = SEPG_CLASS_DB_VIEW;
			tclass_text = "view";
			break;
		default:
			goto out;
	}

	/*
	 * check db_schema:{add_name} permission of the namespace
	 */
	object.classId = NamespaceRelationId;
	object.objectId = classForm->relnamespace;
	object.objectSubId = 0;
	sepgsql_avc_check_perms(&object,
							SEPG_CLASS_DB_SCHEMA,
							SEPG_DB_SCHEMA__ADD_NAME,
							getObjectDescription(&object),
							true);

	/*
	 * Compute a default security label when we create a new relation object
	 * under the specified namespace.
	 */
	scontext = sepgsql_get_client_label();
	tcontext = sepgsql_get_label(NamespaceRelationId,
								 classForm->relnamespace, 0);
	rcontext = sepgsql_compute_create(scontext, tcontext, tclass);

	/*
	 * check db_xxx:{create} permission
	 */
	snprintf(audit_name, sizeof(audit_name), "%s %s",
			 tclass_text, NameStr(classForm->relname));
	sepgsql_avc_check_perms_label(rcontext,
								  tclass,
								  SEPG_DB_DATABASE__CREATE,
								  audit_name,
								  true);

	/*
	 * Assign the default security label on the new relation
	 */
	object.classId = RelationRelationId;
	object.objectId = relOid;
	object.objectSubId = 0;
	SetSecurityLabel(&object, SEPGSQL_LABEL_TAG, rcontext);

//.........这里部分代码省略.........

/*
 * sepgsql_attribute_post_create
 *
 * This routine assigns a default security label on a newly defined
 * column, using ALTER TABLE ... ADD COLUMN.
 * Note that this routine is not invoked in the case of CREATE TABLE,
 * although it also defines columns in addition to table.
 */
void
sepgsql_attribute_post_create(Oid relOid, AttrNumber attnum)
{
	Relation	rel;
	ScanKeyData skey[2];
	SysScanDesc sscan;
	HeapTuple	tuple;
	char	   *scontext;
	char	   *tcontext;
	char	   *ncontext;
	char		audit_name[2 * NAMEDATALEN + 20];
	ObjectAddress object;
	Form_pg_attribute attForm;

	/*
	 * Only attributes within regular relation have individual security
	 * labels.
	 */
	if (get_rel_relkind(relOid) != RELKIND_RELATION)
		return;

	/*
	 * Compute a default security label of the new column underlying the
	 * specified relation, and check permission to create it.
	 */
	rel = heap_open(AttributeRelationId, AccessShareLock);

	ScanKeyInit(&skey[0],
				Anum_pg_attribute_attrelid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(relOid));
	ScanKeyInit(&skey[1],
				Anum_pg_attribute_attnum,
				BTEqualStrategyNumber, F_INT2EQ,
				Int16GetDatum(attnum));

	sscan = systable_beginscan(rel, AttributeRelidNumIndexId, true,
							   SnapshotSelf, 2, &skey[0]);

	tuple = systable_getnext(sscan);
	if (!HeapTupleIsValid(tuple))
		elog(ERROR, "catalog lookup failed for column %d of relation %u",
			 attnum, relOid);

	attForm = (Form_pg_attribute) GETSTRUCT(tuple);

	scontext = sepgsql_get_client_label();
	tcontext = sepgsql_get_label(RelationRelationId, relOid, 0);
	ncontext = sepgsql_compute_create(scontext, tcontext,
									  SEPG_CLASS_DB_COLUMN);

	/*
	 * check db_column:{create} permission
	 */
	snprintf(audit_name, sizeof(audit_name), "table %s column %s",
			 get_rel_name(relOid), NameStr(attForm->attname));
	sepgsql_avc_check_perms_label(ncontext,
								  SEPG_CLASS_DB_COLUMN,
								  SEPG_DB_COLUMN__CREATE,
								  audit_name,
								  true);

	/*
	 * Assign the default security label on a new procedure
	 */
	object.classId = RelationRelationId;
	object.objectId = relOid;
	object.objectSubId = attnum;
	SetSecurityLabel(&object, SEPGSQL_LABEL_TAG, ncontext);

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

	pfree(tcontext);
	pfree(ncontext);
}

/*
 * RemoveRewriteRule
 *
 * Delete a rule given its name.
 */
void
RemoveRewriteRule(Oid owningRel, const char *ruleName, DropBehavior behavior,
				  bool missing_ok)
{
	HeapTuple	tuple;
	Oid			eventRelationOid;
	ObjectAddress object;
	cqContext	*pcqCtx;

	/*
	 * Find the tuple for the target rule.
	 */
	pcqCtx = caql_beginscan(
			NULL,
			cql("SELECT * FROM pg_rewrite "
				" WHERE ev_class = :1 "
				" AND rulename = :2 "
				" FOR UPDATE ",
				ObjectIdGetDatum(owningRel),
				CStringGetDatum((char *) ruleName)));

	tuple = caql_getnext(pcqCtx);

	/*
	 * complain if no rule with such name exists
	 */
	if (!HeapTupleIsValid(tuple))
	{
		if (!missing_ok)
			ereport(ERROR,
					(errcode(ERRCODE_UNDEFINED_OBJECT),
					 errmsg("rule \"%s\" for relation \"%s\" does not exist",
							ruleName, get_rel_name(owningRel))));
		else
			ereport(NOTICE,
					(errmsg("rule \"%s\" for relation \"%s\" does not exist, skipping",
							ruleName, get_rel_name(owningRel))));

		caql_endscan(pcqCtx);
		return;
	}

	/*
	 * Verify user has appropriate permissions.
	 */
	eventRelationOid = ((Form_pg_rewrite) GETSTRUCT(tuple))->ev_class;
	Assert(eventRelationOid == owningRel);
	if (!pg_class_ownercheck(eventRelationOid, GetUserId()))
		aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS,
					   get_rel_name(eventRelationOid));

	/*
	 * Do the deletion
	 */
	object.classId = RewriteRelationId;
	object.objectId = HeapTupleGetOid(tuple);
	object.objectSubId = 0;

	caql_endscan(pcqCtx);

	performDeletion(&object, behavior);
}

/*
 * Fetch parser cache entry
 */
TSParserCacheEntry *
lookup_ts_parser_cache(Oid prsId)
{
	TSParserCacheEntry *entry;

	if (TSParserCacheHash == NULL)
	{
		/* First time through: initialize the hash table */
		HASHCTL		ctl;

		MemSet(&ctl, 0, sizeof(ctl));
		ctl.keysize = sizeof(Oid);
		ctl.entrysize = sizeof(TSParserCacheEntry);
		TSParserCacheHash = hash_create("Tsearch parser cache", 4,
										&ctl, HASH_ELEM | HASH_BLOBS);
		/* Flush cache on pg_ts_parser changes */
		CacheRegisterSyscacheCallback(TSPARSEROID, InvalidateTSCacheCallBack,
									  PointerGetDatum(TSParserCacheHash));

		/* Also make sure CacheMemoryContext exists */
		if (!CacheMemoryContext)
			CreateCacheMemoryContext();
	}

	/* Check single-entry cache */
	if (lastUsedParser && lastUsedParser->prsId == prsId &&
		lastUsedParser->isvalid)
		return lastUsedParser;

	/* Try to look up an existing entry */
	entry = (TSParserCacheEntry *) hash_search(TSParserCacheHash,
											   (void *) &prsId,
											   HASH_FIND, NULL);
	if (entry == NULL || !entry->isvalid)
	{
		/*
		 * If we didn't find one, we want to make one. But first look up the
		 * object to be sure the OID is real.
		 */
		HeapTuple	tp;
		Form_pg_ts_parser prs;

		tp = SearchSysCache1(TSPARSEROID, ObjectIdGetDatum(prsId));
		if (!HeapTupleIsValid(tp))
			elog(ERROR, "cache lookup failed for text search parser %u",
				 prsId);
		prs = (Form_pg_ts_parser) GETSTRUCT(tp);

		/*
		 * Sanity checks
		 */
		if (!OidIsValid(prs->prsstart))
			elog(ERROR, "text search parser %u has no prsstart method", prsId);
		if (!OidIsValid(prs->prstoken))
			elog(ERROR, "text search parser %u has no prstoken method", prsId);
		if (!OidIsValid(prs->prsend))
			elog(ERROR, "text search parser %u has no prsend method", prsId);

		if (entry == NULL)
		{
			bool		found;

			/* Now make the cache entry */
			entry = (TSParserCacheEntry *)
				hash_search(TSParserCacheHash,
							(void *) &prsId,
							HASH_ENTER, &found);
			Assert(!found);		/* it wasn't there a moment ago */
		}

		MemSet(entry, 0, sizeof(TSParserCacheEntry));
		entry->prsId = prsId;
		entry->startOid = prs->prsstart;
		entry->tokenOid = prs->prstoken;
		entry->endOid = prs->prsend;
		entry->headlineOid = prs->prsheadline;
		entry->lextypeOid = prs->prslextype;

		ReleaseSysCache(tp);

		fmgr_info_cxt(entry->startOid, &entry->prsstart, CacheMemoryContext);
		fmgr_info_cxt(entry->tokenOid, &entry->prstoken, CacheMemoryContext);
		fmgr_info_cxt(entry->endOid, &entry->prsend, CacheMemoryContext);
		if (OidIsValid(entry->headlineOid))
			fmgr_info_cxt(entry->headlineOid, &entry->prsheadline,
						  CacheMemoryContext);

		entry->isvalid = true;
	}

	lastUsedParser = entry;

	return entry;
}

/*
 * CheckMyDatabase -- fetch information from the pg_database entry for our DB
 */
static void
CheckMyDatabase(const char *name, bool am_superuser)
{
	HeapTuple	tup;
	Form_pg_database dbform;

	/* Fetch our pg_database row normally, via syscache */
	tup = SearchSysCache(DATABASEOID,
						 ObjectIdGetDatum(MyDatabaseId),
						 0, 0, 0);
	if (!HeapTupleIsValid(tup))
		elog(ERROR, "cache lookup failed for database %u", MyDatabaseId);
	dbform = (Form_pg_database) GETSTRUCT(tup);

	/* This recheck is strictly paranoia */
	if (strcmp(name, NameStr(dbform->datname)) != 0)
		ereport(FATAL,
				(errcode(ERRCODE_UNDEFINED_DATABASE),
				 errmsg("database \"%s\" has disappeared from pg_database",
						name),
				 errdetail("Database OID %u now seems to belong to \"%s\".",
						   MyDatabaseId, NameStr(dbform->datname))));

	/*
	 * Check permissions to connect to the database.
	 *
	 * These checks are not enforced when in standalone mode, so that there is
	 * a way to recover from disabling all access to all databases, for
	 * example "UPDATE pg_database SET datallowconn = false;".
	 *
	 * We do not enforce them for the autovacuum worker processes either.
	 */
	if (IsUnderPostmaster && !IsAutoVacuumWorkerProcess())
	{
		/*
		 * Check that the database is currently allowing connections.
		 */
		if (!dbform->datallowconn)
			ereport(FATAL,
					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
			 errmsg("database \"%s\" is not currently accepting connections",
					name)));

		/*
		 * Check privilege to connect to the database.	(The am_superuser test
		 * is redundant, but since we have the flag, might as well check it
		 * and save a few cycles.)
		 */
		if (!am_superuser &&
			pg_database_aclcheck(MyDatabaseId, GetUserId(),
								 ACL_CONNECT) != ACLCHECK_OK)
			ereport(FATAL,
					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
					 errmsg("permission denied for database \"%s\"", name),
					 errdetail("User does not have CONNECT privilege.")));

		/*
		 * Check connection limit for this database.
		 *
		 * There is a race condition here --- we create our PGPROC before
		 * checking for other PGPROCs.	If two backends did this at about the
		 * same time, they might both think they were over the limit, while
		 * ideally one should succeed and one fail.  Getting that to work
		 * exactly seems more trouble than it is worth, however; instead we
		 * just document that the connection limit is approximate.
		 */
		if (dbform->datconnlimit >= 0 &&
			!am_superuser &&
			CountDBBackends(MyDatabaseId) > dbform->datconnlimit)
			ereport(FATAL,
					(errcode(ERRCODE_TOO_MANY_CONNECTIONS),
					 errmsg("too many connections for database \"%s\"",
							name)));
	}

	/*
	 * OK, we're golden.  Next to-do item is to save the encoding info out of
	 * the pg_database tuple.
	 */
	SetDatabaseEncoding(dbform->encoding);
	/* Record it as a GUC internal option, too */
	SetConfigOption("server_encoding", GetDatabaseEncodingName(),
					PGC_INTERNAL, PGC_S_OVERRIDE);
	/* If we have no other source of client_encoding, use server encoding */
	SetConfigOption("client_encoding", GetDatabaseEncodingName(),
					PGC_BACKEND, PGC_S_DEFAULT);

	/* Use the right encoding in translated messages */
#ifdef ENABLE_NLS
	pg_bind_textdomain_codeset(textdomain(NULL));
#endif

	/*
	 * Lastly, set up any database-specific configuration variables.
	 */
	if (IsUnderPostmaster)
	{
//.........这里部分代码省略.........

//.........这里部分代码省略.........
		InitializeClientEncoding();

		/* report this backend in the PgBackendStatus array */
		pgstat_bestart();

		/* close the transaction we started above */
		CommitTransactionCommand();

		return;
	}

	/*
	 * Set up the global variables holding database id and path.  But note we
	 * won't actually try to touch the database just yet.
	 *
	 * We take a shortcut in the bootstrap case, otherwise we have to look up
	 * the db name in pg_database.
	 */
	if (bootstrap)
	{
		MyDatabaseId = TemplateDbOid;
		MyDatabaseTableSpace = DEFAULTTABLESPACE_OID;
	}
	else if (in_dbname != NULL)
	{
		HeapTuple	tuple;
		Form_pg_database dbform;

		tuple = GetDatabaseTuple(in_dbname);
		if (!HeapTupleIsValid(tuple))
			ereport(FATAL,
					(errcode(ERRCODE_UNDEFINED_DATABASE),
					 errmsg("database \"%s\" does not exist", in_dbname)));
		dbform = (Form_pg_database) GETSTRUCT(tuple);
		MyDatabaseId = HeapTupleGetOid(tuple);
		MyDatabaseTableSpace = dbform->dattablespace;
		/* take database name from the caller, just for paranoia */
		strlcpy(dbname, in_dbname, sizeof(dbname));
		pfree(tuple);
	}
	else
	{
		/* caller specified database by OID */
		HeapTuple	tuple;
		Form_pg_database dbform;

		tuple = GetDatabaseTupleByOid(dboid);
		if (!HeapTupleIsValid(tuple))
			ereport(FATAL,
					(errcode(ERRCODE_UNDEFINED_DATABASE),
					 errmsg("database %u does not exist", dboid)));
		dbform = (Form_pg_database) GETSTRUCT(tuple);
		MyDatabaseId = HeapTupleGetOid(tuple);
		MyDatabaseTableSpace = dbform->dattablespace;
		Assert(MyDatabaseId == dboid);
		strlcpy(dbname, NameStr(dbform->datname), sizeof(dbname));
		/* pass the database name back to the caller */
		if (out_dbname)
			strcpy(out_dbname, dbname);
		pfree(tuple);
	}

	/* Now we can mark our PGPROC entry with the database ID */
	/* (We assume this is an atomic store so no lock is needed) */
	MyProc->databaseId = MyDatabaseId;

/*
 * CheckMyDatabase -- fetch information from the pg_database entry for our DB
 */
static void
CheckMyDatabase(const char *name, bool am_superuser, bool override_allow_connections)
{
	HeapTuple	tup;
	Form_pg_database dbform;
	char	   *collate;
	char	   *ctype;

	/* Fetch our pg_database row normally, via syscache */
	tup = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(MyDatabaseId));
	if (!HeapTupleIsValid(tup))
		elog(ERROR, "cache lookup failed for database %u", MyDatabaseId);
	dbform = (Form_pg_database) GETSTRUCT(tup);

	/* This recheck is strictly paranoia */
	if (strcmp(name, NameStr(dbform->datname)) != 0)
		ereport(FATAL,
				(errcode(ERRCODE_UNDEFINED_DATABASE),
				 errmsg("database \"%s\" has disappeared from pg_database",
						name),
				 errdetail("Database OID %u now seems to belong to \"%s\".",
						   MyDatabaseId, NameStr(dbform->datname))));

	/*
	 * Check permissions to connect to the database.
	 *
	 * These checks are not enforced when in standalone mode, so that there is
	 * a way to recover from disabling all access to all databases, for
	 * example "UPDATE pg_database SET datallowconn = false;".
	 *
	 * We do not enforce them for autovacuum worker processes either.
	 */
	if (IsUnderPostmaster && !IsAutoVacuumWorkerProcess())
	{
		/*
		 * Check that the database is currently allowing connections.
		 */
		if (!dbform->datallowconn && !override_allow_connections)
			ereport(FATAL,
					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
					 errmsg("database \"%s\" is not currently accepting connections",
							name)));

		/*
		 * Check privilege to connect to the database.  (The am_superuser test
		 * is redundant, but since we have the flag, might as well check it
		 * and save a few cycles.)
		 */
		if (!am_superuser &&
			pg_database_aclcheck(MyDatabaseId, GetUserId(),
								 ACL_CONNECT) != ACLCHECK_OK)
			ereport(FATAL,
					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
					 errmsg("permission denied for database \"%s\"", name),
					 errdetail("User does not have CONNECT privilege.")));

		/*
		 * Check connection limit for this database.
		 *
		 * There is a race condition here --- we create our PGPROC before
		 * checking for other PGPROCs.  If two backends did this at about the
		 * same time, they might both think they were over the limit, while
		 * ideally one should succeed and one fail.  Getting that to work
		 * exactly seems more trouble than it is worth, however; instead we
		 * just document that the connection limit is approximate.
		 */
		if (dbform->datconnlimit >= 0 &&
			!am_superuser &&
			CountDBConnections(MyDatabaseId) > dbform->datconnlimit)
			ereport(FATAL,
					(errcode(ERRCODE_TOO_MANY_CONNECTIONS),
					 errmsg("too many connections for database \"%s\"",
							name)));
	}

	/*
	 * OK, we're golden.  Next to-do item is to save the encoding info out of
	 * the pg_database tuple.
	 */
	SetDatabaseEncoding(dbform->encoding);
	/* Record it as a GUC internal option, too */
	SetConfigOption("server_encoding", GetDatabaseEncodingName(),
					PGC_INTERNAL, PGC_S_OVERRIDE);
	/* If we have no other source of client_encoding, use server encoding */
	SetConfigOption("client_encoding", GetDatabaseEncodingName(),
					PGC_BACKEND, PGC_S_DYNAMIC_DEFAULT);

	/* assign locale variables */
	collate = NameStr(dbform->datcollate);
	ctype = NameStr(dbform->datctype);

	if (pg_perm_setlocale(LC_COLLATE, collate) == NULL)
		ereport(FATAL,
				(errmsg("database locale is incompatible with operating system"),
				 errdetail("The database was initialized with LC_COLLATE \"%s\", "
						   " which is not recognized by setlocale().", collate),
				 errhint("Recreate the database with another locale or install the missing locale.")));
//.........这里部分代码省略.........

//.........这里部分代码省略.........
		InitializeClientEncoding();

		/* report this backend in the PgBackendStatus array */
		pgstat_bestart();

		/* close the transaction we started above */
		CommitTransactionCommand();

		return;
	}

	/*
	 * Set up the global variables holding database id and default tablespace.
	 * But note we won't actually try to touch the database just yet.
	 *
	 * We take a shortcut in the bootstrap case, otherwise we have to look up
	 * the db's entry in pg_database.
	 */
	if (bootstrap)
	{
		MyDatabaseId = TemplateDbOid;
		MyDatabaseTableSpace = DEFAULTTABLESPACE_OID;
	}
	else if (in_dbname != NULL)
	{
		HeapTuple	tuple;
		Form_pg_database dbform;

		tuple = GetDatabaseTuple(in_dbname);
		if (!HeapTupleIsValid(tuple))
			ereport(FATAL,
					(errcode(ERRCODE_UNDEFINED_DATABASE),
					 errmsg("database \"%s\" does not exist", in_dbname)));
		dbform = (Form_pg_database) GETSTRUCT(tuple);
		MyDatabaseId = dbform->oid;
		MyDatabaseTableSpace = dbform->dattablespace;
		/* take database name from the caller, just for paranoia */
		strlcpy(dbname, in_dbname, sizeof(dbname));
	}
	else if (OidIsValid(dboid))
	{
		/* caller specified database by OID */
		HeapTuple	tuple;
		Form_pg_database dbform;

		tuple = GetDatabaseTupleByOid(dboid);
		if (!HeapTupleIsValid(tuple))
			ereport(FATAL,
					(errcode(ERRCODE_UNDEFINED_DATABASE),
					 errmsg("database %u does not exist", dboid)));
		dbform = (Form_pg_database) GETSTRUCT(tuple);
		MyDatabaseId = dbform->oid;
		MyDatabaseTableSpace = dbform->dattablespace;
		Assert(MyDatabaseId == dboid);
		strlcpy(dbname, NameStr(dbform->datname), sizeof(dbname));
		/* pass the database name back to the caller */
		if (out_dbname)
			strcpy(out_dbname, dbname);
	}
	else
	{
		/*
		 * If this is a background worker not bound to any particular
		 * database, we're done now.  Everything that follows only makes sense
		 * if we are bound to a specific database.  We do need to close the
		 * transaction we started before returning.

static PyObj
func_new_from_oid(PyTypeObject *subtype, Oid fn_oid, PyObj fn_oid_int, PyObj fn_oid_str)
{
	volatile HeapTuple ht = NULL;
	volatile PyObj rob = NULL;

	Assert(OidIsValid(fn_oid));
	Assert(fn_oid_int != NULL);
	Assert(fn_oid_str != NULL);

	rob = subtype->tp_alloc(subtype, 0);
	if (rob == NULL)
		return(NULL);

	PyPgFunction_SetOid(rob, fn_oid);
	PyPgFunction_SetStateful(rob, false);

	Py_INCREF(fn_oid_int);
	Py_INCREF(fn_oid_str);

	PyPgFunction_SetPyLongOid(rob, fn_oid_int);
	PyPgFunction_SetPyUnicodeOid(rob, fn_oid_str);

	/*
	 * Collect the Function information from the system cache
	 */
	PG_TRY();
	{
		Form_pg_proc ps;
		Form_pg_namespace ns;
		FmgrInfo flinfo;
		text *prosrc;
		Datum prosrc_datum;
		bool isnull = true;
		const char *filename = NULL, *nspname, *q_nspname;
		TupleDesc argdesc = NULL, result_desc = NULL;
		Oid prorettype = InvalidOid;
		PyObj id_str_ob = NULL, nspname_str_ob = NULL;
		PyObj filename_str_ob = NULL, q_nspname_str_ob = NULL;
		PyObj output = NULL, src = NULL;
		PyObj input;

		ht = SearchSysCache(PROCOID, fn_oid, 0, 0, 0);
		if (!HeapTupleIsValid(ht))
		{
			ereport(ERROR,(
				errcode(ERRCODE_UNDEFINED_FUNCTION),
				errmsg("failed to find function at oid %d", fn_oid)
			));
		}

		PyPgFunction_SetXMin(rob, HeapTupleHeaderGetXmin(ht->t_data));
		PyPgFunction_SetItemPointer(rob, &(ht->t_self));

		ps = (Form_pg_proc) GETSTRUCT(ht);

		PyPgFunction_SetNamespace(rob, ps->pronamespace);
		PyPgFunction_SetLanguage(rob, ps->prolang);
		PyPgFunction_SetReturnsSet(rob, ps->proretset);
		PyPgFunction_SetVolatile(rob, ps->provolatile);

		prorettype = ps->prorettype;
		prosrc_datum = SysCacheGetAttr(
			PROCOID, ht, Anum_pg_proc_prosrc, &isnull);
		if (!isnull)
		{
			prosrc = DatumGetTextPCopy(prosrc_datum);
			src = PyUnicode_FromTEXT(prosrc);
			PyPgFunction_SetSource(rob, src);
			pfree(prosrc);
			prosrc = NULL;
		}
		else
		{
			src = Py_None;
			Py_INCREF(src);
			PyPgFunction_SetSource(rob, src);
		}
		if (src == NULL)
			PyErr_RelayException();

		/*
		 * Get the function's address.
		 */
		fmgr_info(fn_oid, &flinfo);
		PyPgFunction_SetPGFunction(rob, flinfo.fn_addr);

		/*
		 * Build function parameters TupleDesc
		 */
		if (ps->pronargs > 0)
		{
			argdesc = TupleDesc_From_pg_proc_arginfo(ht);
			input = PyPgTupleDesc_FromCopy(argdesc);
			if (input == NULL)
				PyErr_RelayException();
			PyPgFunction_SetInput(rob, input);
			FreeTupleDesc(argdesc);
		}
		else
//.........这里部分代码省略.........

/*
 * sepgsql_relation_post_create
 *
 * The post creation hook of relation/attribute
 */
void
sepgsql_relation_post_create(Oid relOid)
{
	Relation	rel;
	ScanKeyData skey;
	SysScanDesc sscan;
	HeapTuple	tuple;
	Form_pg_class classForm;
	ObjectAddress object;
	uint16		tclass;
	char	   *scontext;		/* subject */
	char	   *tcontext;		/* schema */
	char	   *rcontext;		/* relation */
	char	   *ccontext;		/* column */

	/*
	 * Fetch catalog record of the new relation. Because pg_class entry is not
	 * visible right now, we need to scan the catalog using SnapshotSelf.
	 */
	rel = heap_open(RelationRelationId, AccessShareLock);

	ScanKeyInit(&skey,
				ObjectIdAttributeNumber,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(relOid));

	sscan = systable_beginscan(rel, ClassOidIndexId, true,
							   SnapshotSelf, 1, &skey);

	tuple = systable_getnext(sscan);
	if (!HeapTupleIsValid(tuple))
		elog(ERROR, "catalog lookup failed for relation %u", relOid);

	classForm = (Form_pg_class) GETSTRUCT(tuple);

	if (classForm->relkind == RELKIND_RELATION)
		tclass = SEPG_CLASS_DB_TABLE;
	else if (classForm->relkind == RELKIND_SEQUENCE)
		tclass = SEPG_CLASS_DB_SEQUENCE;
	else if (classForm->relkind == RELKIND_VIEW)
		tclass = SEPG_CLASS_DB_VIEW;
	else
		goto out;				/* No need to assign individual labels */

	/*
	 * Compute a default security label when we create a new relation object
	 * under the specified namespace.
	 */
	scontext = sepgsql_get_client_label();
	tcontext = sepgsql_get_label(NamespaceRelationId,
								 classForm->relnamespace, 0);
	rcontext = sepgsql_compute_create(scontext, tcontext, tclass);

	/*
	 * Assign the default security label on the new relation
	 */
	object.classId = RelationRelationId;
	object.objectId = relOid;
	object.objectSubId = 0;
	SetSecurityLabel(&object, SEPGSQL_LABEL_TAG, rcontext);

	/*
	 * We also assigns a default security label on columns of the new regular
	 * tables.
	 */
	if (classForm->relkind == RELKIND_RELATION)
	{
		AttrNumber	index;

		ccontext = sepgsql_compute_create(scontext, rcontext,
										  SEPG_CLASS_DB_COLUMN);
		for (index = FirstLowInvalidHeapAttributeNumber + 1;
			 index <= classForm->relnatts;
			 index++)
		{
			if (index == InvalidAttrNumber)
				continue;

			if (index == ObjectIdAttributeNumber && !classForm->relhasoids)
				continue;

			object.classId = RelationRelationId;
			object.objectId = relOid;
			object.objectSubId = index;
			SetSecurityLabel(&object, SEPGSQL_LABEL_TAG, ccontext);
		}
		pfree(ccontext);
	}
	pfree(rcontext);
out:
	systable_endscan(sscan);
	heap_close(rel, AccessShareLock);
}

static int64
PersistentBuild_TruncateAllGpRelationNode(void)
{
	Relation pg_database;
	HeapScanDesc scan;
	HeapTuple tuple;

	int64 count;

	pg_database = heap_open(
						DatabaseRelationId,
						AccessShareLock);

	/*
	 * Truncate gp_relation_node and its index in each database.
	 */
	scan = heap_beginscan(pg_database, SnapshotNow, 0, NULL);
	count = 0;
	while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
	{
		Form_pg_database form_pg_database =
						(Form_pg_database)GETSTRUCT(tuple);

		Oid dbOid;
		Oid dattablespace;
		RelFileNode relFileNode;
		SMgrRelation smgrRelation;
		Page btree_metapage;
		
		dbOid = HeapTupleGetOid(tuple);
		dattablespace = form_pg_database->dattablespace;

		if (dbOid == HcatalogDbOid)
			continue;

		if (Debug_persistent_print)
			elog(Persistent_DebugPrintLevel(), 
				 "PersistentBuild_TruncateAllGpRelationNode: dbOid %u, '%s'",
				 dbOid,
				 form_pg_database->datname.data);

		if (Debug_persistent_print)
			elog(Persistent_DebugPrintLevel(), 
				 "Truncating gp_relation_node %u/%u/%u in database oid %u ('%s')",
				 relFileNode.spcNode,
				 relFileNode.dbNode,
				 relFileNode.relNode,
				 dbOid,
				 form_pg_database->datname.data);

		relFileNode.spcNode = dattablespace;
		relFileNode.dbNode = dbOid;
		relFileNode.relNode = GpRelfileNodeRelationId;

		/*
		 * Truncate WITHOUT generating an XLOG record (i.e. pretend it is a temp relation).
		 */
		PersistentBuild_NonTransactionTruncate(&relFileNode);
		count++;

		/*
		 * And, the index.  Unfortunately, the relfilenode OID can change due to a
		 * REINDEX {TABLE|INDEX} command.
		 */
		PersistentBuild_FindGpRelationNodeIndex(
											dbOid,
											dattablespace,
											&relFileNode);

		if (Debug_persistent_print)
			elog(Persistent_DebugPrintLevel(), 
				 "Truncating gp_relation_node_index %u/%u/%u in database oid %u ('%s').  relfilenode different %s, tablespace different %s",
				 relFileNode.spcNode,
				 relFileNode.dbNode,
				 relFileNode.relNode,
				 dbOid,
				 form_pg_database->datname.data,
				 ((relFileNode.relNode != GpRelfileNodeOidIndexId) ? "true" : "false"),
				 ((relFileNode.spcNode != dattablespace) ? "true" : "false"));

		PersistentBuild_NonTransactionTruncate(&relFileNode);

		// The BTree needs an empty meta-data block.
		smgrRelation = smgropen(relFileNode);

		btree_metapage = (Page)palloc(BLCKSZ);
		_bt_initmetapage(btree_metapage, P_NONE, 0);
		smgrwrite(
			smgrRelation, 
			/* blockNum */ 0, 
			(char*)btree_metapage,
			/* isTemp */ false);
		smgrimmedsync(smgrRelation);
		pfree(btree_metapage);

		smgrclose(smgrRelation);

		count++;
	}

//.........这里部分代码省略.........

/*
 * AggregateCreate
 */
void
AggregateCreate(const char *aggName,
				Oid aggNamespace,
				Oid *aggArgTypes,
				int numArgs,
				List *aggtransfnName,
				List *aggfinalfnName,
				List *aggsortopName,
				Oid aggTransType,
				const char *agginitval)
{
	Relation	aggdesc;
	HeapTuple	tup;
	bool		nulls[Natts_pg_aggregate];
	Datum		values[Natts_pg_aggregate];
	Form_pg_proc proc;
	Oid			transfn;
	Oid			finalfn = InvalidOid;	/* can be omitted */
	Oid			sortop = InvalidOid;	/* can be omitted */
	bool		hasPolyArg;
	bool		hasInternalArg;
	Oid			rettype;
	Oid			finaltype;
	Oid		   *fnArgs;
	int			nargs_transfn;
	Oid			procOid;
	TupleDesc	tupDesc;
	int			i;
	ObjectAddress myself,
				referenced;

	/* sanity checks (caller should have caught these) */
	if (!aggName)
		elog(ERROR, "no aggregate name supplied");

	if (!aggtransfnName)
		elog(ERROR, "aggregate must have a transition function");

	/* check for polymorphic and INTERNAL arguments */
	hasPolyArg = false;
	hasInternalArg = false;
	for (i = 0; i < numArgs; i++)
	{
		if (IsPolymorphicType(aggArgTypes[i]))
			hasPolyArg = true;
		else if (aggArgTypes[i] == INTERNALOID)
			hasInternalArg = true;
	}

	/*
	 * If transtype is polymorphic, must have polymorphic argument also; else
	 * we will have no way to deduce the actual transtype.
	 */
	if (IsPolymorphicType(aggTransType) && !hasPolyArg)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_FUNCTION_DEFINITION),
				 errmsg("cannot determine transition data type"),
				 errdetail("An aggregate using a polymorphic transition type must have at least one polymorphic argument.")));

	/* find the transfn */
	nargs_transfn = numArgs + 1;
	fnArgs = (Oid *) palloc(nargs_transfn * sizeof(Oid));
	fnArgs[0] = aggTransType;
	memcpy(fnArgs + 1, aggArgTypes, numArgs * sizeof(Oid));
	transfn = lookup_agg_function(aggtransfnName, nargs_transfn, fnArgs,
								  &rettype);

	/*
	 * Return type of transfn (possibly after refinement by
	 * enforce_generic_type_consistency, if transtype isn't polymorphic) must
	 * exactly match declared transtype.
	 *
	 * In the non-polymorphic-transtype case, it might be okay to allow a
	 * rettype that's binary-coercible to transtype, but I'm not quite
	 * convinced that it's either safe or useful.  When transtype is
	 * polymorphic we *must* demand exact equality.
	 */
	if (rettype != aggTransType)
		ereport(ERROR,
				(errcode(ERRCODE_DATATYPE_MISMATCH),
				 errmsg("return type of transition function %s is not %s",
						NameListToString(aggtransfnName),
						format_type_be(aggTransType))));

	tup = SearchSysCache1(PROCOID, ObjectIdGetDatum(transfn));
	if (!HeapTupleIsValid(tup))
		elog(ERROR, "cache lookup failed for function %u", transfn);
	proc = (Form_pg_proc) GETSTRUCT(tup);

	/*
	 * If the transfn is strict and the initval is NULL, make sure first input
	 * type and transtype are the same (or at least binary-compatible), so
	 * that it's OK to use the first input value as the initial transValue.
	 */
	if (proc->proisstrict && agginitval == NULL)
	{
		if (numArgs < 1 ||
//.........这里部分代码省略.........

/*
 * Rename a tablespace
 */
ObjectAddress
RenameTableSpace(const char *oldname, const char *newname)
{
	Oid			tspId;
	Relation	rel;
	ScanKeyData entry[1];
	HeapScanDesc scan;
	HeapTuple	tup;
	HeapTuple	newtuple;
	Form_pg_tablespace newform;
	ObjectAddress address;

	/* Search pg_tablespace */
	rel = heap_open(TableSpaceRelationId, RowExclusiveLock);

	ScanKeyInit(&entry[0],
				Anum_pg_tablespace_spcname,
				BTEqualStrategyNumber, F_NAMEEQ,
				CStringGetDatum(oldname));
	scan = heap_beginscan_catalog(rel, 1, entry);
	tup = heap_getnext(scan, ForwardScanDirection);
	if (!HeapTupleIsValid(tup))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("tablespace \"%s\" does not exist",
						oldname)));

	tspId = HeapTupleGetOid(tup);
	newtuple = heap_copytuple(tup);
	newform = (Form_pg_tablespace) GETSTRUCT(newtuple);

	heap_endscan(scan);

	/* Must be owner */
	if (!pg_tablespace_ownercheck(HeapTupleGetOid(newtuple), GetUserId()))
		aclcheck_error(ACLCHECK_NO_PRIV, ACL_KIND_TABLESPACE, oldname);

	/* Validate new name */
	if (!allowSystemTableMods && IsReservedName(newname))
		ereport(ERROR,
				(errcode(ERRCODE_RESERVED_NAME),
				 errmsg("unacceptable tablespace name \"%s\"", newname),
		errdetail("The prefix \"pg_\" is reserved for system tablespaces.")));

	/* Make sure the new name doesn't exist */
	ScanKeyInit(&entry[0],
				Anum_pg_tablespace_spcname,
				BTEqualStrategyNumber, F_NAMEEQ,
				CStringGetDatum(newname));
	scan = heap_beginscan_catalog(rel, 1, entry);
	tup = heap_getnext(scan, ForwardScanDirection);
	if (HeapTupleIsValid(tup))
		ereport(ERROR,
				(errcode(ERRCODE_DUPLICATE_OBJECT),
				 errmsg("tablespace \"%s\" already exists",
						newname)));

	heap_endscan(scan);

	/* OK, update the entry */
	namestrcpy(&(newform->spcname), newname);

	simple_heap_update(rel, &newtuple->t_self, newtuple);
	CatalogUpdateIndexes(rel, newtuple);

	InvokeObjectPostAlterHook(TableSpaceRelationId, tspId, 0);

	ObjectAddressSet(address, TableSpaceRelationId, tspId);

	heap_close(rel, NoLock);

	return address;
}

/*
 * TupleDescInitEntry
 *		This function initializes a single attribute structure in
 *		a previously allocated tuple descriptor.
 */
void
TupleDescInitEntry(TupleDesc desc,
				   AttrNumber attributeNumber,
				   const char *attributeName,
				   Oid oidtypeid,
				   int32 typmod,
				   int attdim)
{
	HeapTuple	tuple;
	Form_pg_type typeForm;
	Form_pg_attribute att;

	/*
	 * sanity checks
	 */
	AssertArg(PointerIsValid(desc));
	AssertArg(attributeNumber >= 1);
	AssertArg(attributeNumber <= desc->natts);

	/*
	 * initialize the attribute fields
	 */
	att = desc->attrs[attributeNumber - 1];

	att->attrelid = 0;			/* dummy value */

	/*
	 * Note: attributeName can be NULL, because the planner doesn't always
	 * fill in valid resname values in targetlists, particularly for resjunk
	 * attributes.
	 */
	if (attributeName != NULL)
		namestrcpy(&(att->attname), attributeName);
	else
		MemSet(NameStr(att->attname), 0, NAMEDATALEN);

	att->attstattarget = -1;
	att->attcacheoff = -1;
	att->atttypmod = typmod;

	att->attnum = attributeNumber;
	att->attndims = attdim;

	att->