/*
 * OperatorShellMake
 *		Make a "shell" entry for a not-yet-existing operator.
 */
static Oid
OperatorShellMake(const char *operatorName,
				  Oid operatorNamespace,
				  Oid leftTypeId,
				  Oid rightTypeId)
{
	Relation	pg_operator_desc;
	Oid			operatorObjectId;
	int			i;
	HeapTuple	tup;
	Datum		values[Natts_pg_operator];
	char		nulls[Natts_pg_operator];
	NameData	oname;
	TupleDesc	tupDesc;

	/*
	 * validate operator name
	 */
	if (!validOperatorName(operatorName))
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_NAME),
				 errmsg("\"%s\" is not a valid operator name",
						operatorName)));

	/*
	 * initialize our *nulls and *values arrays
	 */
	for (i = 0; i < Natts_pg_operator; ++i)
	{
		nulls[i] = ' ';
		values[i] = (Datum) NULL;		/* redundant, but safe */
	}

	/*
	 * initialize values[] with the operator name and input data types.
	 * Note that oprcode is set to InvalidOid, indicating it's a shell.
	 */
	i = 0;
	namestrcpy(&oname, operatorName);
	values[i++] = NameGetDatum(&oname); /* oprname */
	values[i++] = ObjectIdGetDatum(operatorNamespace);	/* oprnamespace */
	values[i++] = Int32GetDatum(GetUserId());	/* oprowner */
	values[i++] = CharGetDatum(leftTypeId ? (rightTypeId ? 'b' : 'r') : 'l');	/* oprkind */
	values[i++] = BoolGetDatum(false);	/* oprcanhash */
	values[i++] = ObjectIdGetDatum(leftTypeId); /* oprleft */
	values[i++] = ObjectIdGetDatum(rightTypeId);		/* oprright */
	values[i++] = ObjectIdGetDatum(InvalidOid); /* oprresult */
	values[i++] = ObjectIdGetDatum(InvalidOid); /* oprcom */
	values[i++] = ObjectIdGetDatum(InvalidOid); /* oprnegate */
	values[i++] = ObjectIdGetDatum(InvalidOid); /* oprlsortop */
	values[i++] = ObjectIdGetDatum(InvalidOid); /* oprrsortop */
	values[i++] = ObjectIdGetDatum(InvalidOid); /* oprltcmpop */
	values[i++] = ObjectIdGetDatum(InvalidOid); /* oprgtcmpop */
	values[i++] = ObjectIdGetDatum(InvalidOid); /* oprcode */
	values[i++] = ObjectIdGetDatum(InvalidOid); /* oprrest */
	values[i++] = ObjectIdGetDatum(InvalidOid); /* oprjoin */

	/*
	 * open pg_operator
	 */
	pg_operator_desc = heap_openr(OperatorRelationName, RowExclusiveLock);
	tupDesc = pg_operator_desc->rd_att;

	/*
	 * create a new operator tuple
	 */
	tup = heap_formtuple(tupDesc, values, nulls);

	/*
	 * insert our "shell" operator tuple
	 */
	operatorObjectId = simple_heap_insert(pg_operator_desc, tup);

	CatalogUpdateIndexes(pg_operator_desc, tup);

	/* Add dependencies for the entry */
	makeOperatorDependencies(tup, RelationGetRelid(pg_operator_desc));

	heap_freetuple(tup);

	/*
	 * close the operator relation and return the oid.
	 */
	heap_close(pg_operator_desc, RowExclusiveLock);

	return operatorObjectId;
}

/*
 * OperatorUpd
 *
 *	For a given operator, look up its negator and commutator operators.
 *	If they are defined, but their negator and commutator fields
 *	(respectively) are empty, then use the new operator for neg or comm.
 *	This solves a problem for users who need to insert two new operators
 *	which are the negator or commutator of each other.
 */
static void
OperatorUpd(Oid baseId, Oid commId, Oid negId)
{
	int			i;
	Relation	pg_operator_desc;
	HeapTuple	tup;
	char		nulls[Natts_pg_operator];
	char		replaces[Natts_pg_operator];
	Datum		values[Natts_pg_operator];

	for (i = 0; i < Natts_pg_operator; ++i)
	{
		values[i] = (Datum) 0;
		replaces[i] = ' ';
		nulls[i] = ' ';
	}

	/*
	 * check and update the commutator & negator, if necessary
	 *
	 * First make sure we can see them...
	 */
	CommandCounterIncrement();

	pg_operator_desc = heap_openr(OperatorRelationName, RowExclusiveLock);

	tup = SearchSysCacheCopy(OPEROID,
							 ObjectIdGetDatum(commId),
							 0, 0, 0);

	/*
	 * if the commutator and negator are the same operator, do one update.
	 * XXX this is probably useless code --- I doubt it ever makes sense
	 * for commutator and negator to be the same thing...
	 */
	if (commId == negId)
	{
		if (HeapTupleIsValid(tup))
		{
			Form_pg_operator t = (Form_pg_operator) GETSTRUCT(tup);

			if (!OidIsValid(t->oprcom) || !OidIsValid(t->oprnegate))
			{
				if (!OidIsValid(t->oprnegate))
				{
					values[Anum_pg_operator_oprnegate - 1] = ObjectIdGetDatum(baseId);
					replaces[Anum_pg_operator_oprnegate - 1] = 'r';
				}

				if (!OidIsValid(t->oprcom))
				{
					values[Anum_pg_operator_oprcom - 1] = ObjectIdGetDatum(baseId);
					replaces[Anum_pg_operator_oprcom - 1] = 'r';
				}

				tup = heap_modifytuple(tup,
									   pg_operator_desc,
									   values,
									   nulls,
									   replaces);

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

				CatalogUpdateIndexes(pg_operator_desc, tup);
			}
		}

		heap_close(pg_operator_desc, RowExclusiveLock);

		return;
	}

	/* if commutator and negator are different, do two updates */

	if (HeapTupleIsValid(tup) &&
		!(OidIsValid(((Form_pg_operator) GETSTRUCT(tup))->oprcom)))
	{
		values[Anum_pg_operator_oprcom - 1] = ObjectIdGetDatum(baseId);
		replaces[Anum_pg_operator_oprcom - 1] = 'r';

		tup = heap_modifytuple(tup,
							   pg_operator_desc,
							   values,
							   nulls,
							   replaces);

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

		CatalogUpdateIndexes(pg_operator_desc, tup);

		values[Anum_pg_operator_oprcom - 1] = (Datum) NULL;
//.........这里部分代码省略.........

Datum serialize_array(PG_FUNCTION_ARGS)
{
	ArrayType  *v = PG_GETARG_ARRAYTYPE_P(0);
	Oid		 element_type = ARR_ELEMTYPE(v);
	int16		typlen;
	bool		typbyval;
	char		typalign;
	char		typdelim;
	char	   *p, *value;
	char 		type_category;
	bool		needComma = false;

	bits8	  *bitmap;
	int		 bitmask;
	int		 nitems, i;
	int		 ndim, *dims;

	Oid			typioparam, typiofunc;
	HeapTuple 	type_tuple;
	FmgrInfo	proc, flinfo;

	StringInfoData buf;

//	FILE* log;

//	log = fopen("/var/lib/postgresql/serializer.log", "a");

//	fprintf(log, "Doing serialize_array\n");
//	fflush(log);

	/*
	 * Get info about element type, including its output conversion proc
	 */
	get_type_io_data(element_type, IOFunc_output,
					 &typlen, &typbyval,
					 &typalign, &typdelim,
					 &typioparam, &typiofunc);

	fmgr_info_cxt( typiofunc, &proc, fcinfo->flinfo->fn_mcxt );

	ndim = ARR_NDIM(v);
	dims = ARR_DIMS(v);
	nitems = ArrayGetNItems(ndim, dims);

	if( ndim > 1 )
		elog( ERROR, "multidimensional arrays doesn't supported" );

	p = ARR_DATA_PTR(v);
	bitmap = ARR_NULLBITMAP(v);
	bitmask = 1;

	/* obtain type information from pg_catalog */
	type_tuple = SearchSysCache1( TYPEOID, ObjectIdGetDatum(element_type) );
	if (!HeapTupleIsValid( type_tuple ))
		elog(ERROR, "cache lookup failed for relation %u", element_type);

	type_category = ((Form_pg_type) GETSTRUCT( type_tuple ))->typcategory;

	ReleaseSysCache( type_tuple );

	/* Build the result string */
	initStringInfo(&buf);

	appendStringInfoChar(&buf, '[');
	for (i = 0; i < nitems; i++)
	{
		if (needComma)
			appendStringInfoChar(&buf, ',');
		needComma = true;

		/* Get source element, checking for NULL */
		if (bitmap && (*bitmap & bitmask) == 0)
		{
			// append null
			appendStringInfoString(&buf, "null");
		}
		else
		{
			/* get item value and advance array data pointer */
			Datum itemvalue = fetch_att(p, typbyval, typlen);
			p = att_addlength_pointer(p, typlen, p);
			p = (char *) att_align_nominal(p, typalign);

			//------------------------
			switch( type_category )
			{
				// http://www.postgresql.org/docs/current/static/catalog-pg-type.html#CATALOG-TYPCATEGORY-TABLE

				case 'A': //array - impossible case
				break;

				case 'C': //composite
					//call to serialize_record( ... )
					MemSet( &flinfo, 0, sizeof( flinfo ) );
					flinfo.fn_addr = serialize_record;
					flinfo.fn_nargs = 1;
					flinfo.fn_mcxt = fcinfo->flinfo->fn_mcxt;

					value = PG_TEXT_DATUM_GET_CSTR( FunctionCall1( &flinfo, itemvalue ) );

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

/*
 * regoperout		- converts operator OID to "opr_name"
 */
Datum
regoperout(PG_FUNCTION_ARGS)
{
	Oid			oprid = PG_GETARG_OID(0);
	char	   *result;
	HeapTuple	opertup;

	if (oprid == InvalidOid)
	{
		result = pstrdup("0");
		PG_RETURN_CSTRING(result);
	}

	opertup = SearchSysCache(OPEROID,
							 ObjectIdGetDatum(oprid),
							 0, 0, 0);

	if (HeapTupleIsValid(opertup))
	{
		Form_pg_operator operform = (Form_pg_operator) GETSTRUCT(opertup);
		char	   *oprname = NameStr(operform->oprname);

		/*
		 * In bootstrap mode, skip the fancy namespace stuff and just return
		 * the oper name.  (This path is only needed for debugging output
		 * anyway.)
		 */
		if (IsBootstrapProcessingMode())
			result = pstrdup(oprname);
		else
		{
			FuncCandidateList clist;

			/*
			 * Would this oper be found (uniquely!) by regoperin? If not,
			 * qualify it.
			 */
			clist = OpernameGetCandidates(list_make1(makeString(oprname)),
										  '\0');
			if (clist != NULL && clist->next == NULL &&
				clist->oid == oprid)
				result = pstrdup(oprname);
			else
			{
				const char *nspname;

				nspname = get_namespace_name(operform->oprnamespace);
				nspname = quote_identifier(nspname);
				result = (char *) palloc(strlen(nspname) + strlen(oprname) + 2);
				sprintf(result, "%s.%s", nspname, oprname);
			}
		}

		ReleaseSysCache(opertup);
	}
	else
	{
		/*
		 * If OID doesn't match any pg_operator entry, return it numerically
		 */
		result = (char *) palloc(NAMEDATALEN);
		snprintf(result, NAMEDATALEN, "%u", oprid);
	}

	PG_RETURN_CSTRING(result);
}

/*
 * CreateSharedComments --
 *
 * Create a comment for the specified shared object descriptor.  Inserts a
 * new pg_shdescription tuple, or replaces an existing one with the same key.
 *
 * If the comment given is null or an empty string, instead delete any
 * existing comment for the specified key.
 */
void
CreateSharedComments(Oid oid, Oid classoid, char *comment)
{
	Relation	shdescription;
	ScanKeyData skey[2];
	SysScanDesc sd;
	HeapTuple	oldtuple;
	HeapTuple	newtuple = NULL;
	Datum		values[Natts_pg_shdescription];
	bool		nulls[Natts_pg_shdescription];
	bool		replaces[Natts_pg_shdescription];
	int			i;

	/* Reduce empty-string to NULL case */
	if (comment != NULL && strlen(comment) == 0)
		comment = NULL;

	/* Prepare to form or update a tuple, if necessary */
	if (comment != NULL)
	{
		for (i = 0; i < Natts_pg_shdescription; i++)
		{
			nulls[i] = false;
			replaces[i] = true;
		}
		values[Anum_pg_shdescription_objoid - 1] = ObjectIdGetDatum(oid);
		values[Anum_pg_shdescription_classoid - 1] = ObjectIdGetDatum(classoid);
		values[Anum_pg_shdescription_description - 1] = CStringGetTextDatum(comment);
	}

	/* Use the index to search for a matching old tuple */

	ScanKeyInit(&skey[0],
				Anum_pg_shdescription_objoid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(oid));
	ScanKeyInit(&skey[1],
				Anum_pg_shdescription_classoid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(classoid));

	shdescription = heap_open(SharedDescriptionRelationId, RowExclusiveLock);

	sd = systable_beginscan(shdescription, SharedDescriptionObjIndexId, true,
							SnapshotNow, 2, skey);

	while ((oldtuple = systable_getnext(sd)) != NULL)
	{
		/* Found the old tuple, so delete or update it */

		if (comment == NULL)
			simple_heap_delete(shdescription, &oldtuple->t_self);
		else
		{
			newtuple = heap_modify_tuple(oldtuple, RelationGetDescr(shdescription),
										 values, nulls, replaces);
			simple_heap_update(shdescription, &oldtuple->t_self, newtuple);
		}

		break;					/* Assume there can be only one match */
	}

	systable_endscan(sd);

	/* If we didn't find an old tuple, insert a new one */

	if (newtuple == NULL && comment != NULL)
	{
		newtuple = heap_form_tuple(RelationGetDescr(shdescription),
								   values, nulls);
		simple_heap_insert(shdescription, newtuple);
	}

	/* Update indexes, if necessary */
	if (newtuple != NULL)
	{
		CatalogUpdateIndexes(shdescription, newtuple);
		heap_freetuple(newtuple);
	}

	/* Done */

	heap_close(shdescription, NoLock);
}

/*
 * 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,
		   Oid ruleOid)
{
	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;
	HeapTuple	tup,
				oldtup;
	Oid			rewriteObjectId;
	ObjectAddress myself,
				referenced;
	bool		is_update = false;
	cqContext  *pcqCtx;

	/*
	 * 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++] = BoolGetDatum(evinstead);		/* is_instead */
	values[i++] = CStringGetTextDatum(evqual);	/* ev_qual */
	values[i++] = CStringGetTextDatum(actiontree);		/* ev_action */

	/*
	 * Ready to store new pg_rewrite tuple
	 */

	/*
	 * Check to see if we are replacing an existing tuple
	 */
	pcqCtx = caql_beginscan(
			NULL,
			cql("SELECT * FROM pg_rewrite "
				" WHERE ev_class = :1 "
				" AND rulename = :2 "
				" FOR UPDATE ",
				ObjectIdGetDatum(eventrel_oid),
				CStringGetDatum(rulname)));
	
	oldtup = caql_getnext(pcqCtx);

	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 = caql_modify_current(pcqCtx,
								  values, nulls, replaces);

		caql_update_current(pcqCtx, tup);
		/* and Update indexes (implicit) */

		rewriteObjectId = HeapTupleGetOid(tup);
		is_update = true;
	}
	else
	{
		tup = caql_form_tuple(pcqCtx, values, nulls);

		if (OidIsValid(ruleOid))
			HeapTupleSetOid(tup, ruleOid);

		rewriteObjectId = caql_insert(pcqCtx, tup);
		/* and Update indexes (implicit) */
//.........这里部分代码省略.........

//.........这里部分代码省略.........
			LWLockAcquire(FirstBufMappingLock + i, LW_SHARED);

		/*
		 * Scan though all the buffers, saving the relevant fields in the
		 * fctx->record structure.
		 */
		for (i = 0, bufHdr = BufferDescriptors; i < NBuffers; i++, bufHdr++)
		{
			/* Lock each buffer header before inspecting. */
			LockBufHdr(bufHdr);

			fctx->record[i].bufferid = BufferDescriptorGetBuffer(bufHdr);
			fctx->record[i].relfilenode = bufHdr->tag.rnode.relNode;
			fctx->record[i].reltablespace = bufHdr->tag.rnode.spcNode;
			fctx->record[i].reldatabase = bufHdr->tag.rnode.dbNode;
			fctx->record[i].forknum = bufHdr->tag.forkNum;
			fctx->record[i].blocknum = bufHdr->tag.blockNum;
			fctx->record[i].usagecount = bufHdr->usage_count;

			if (bufHdr->flags & BM_DIRTY)
				fctx->record[i].isdirty = true;
			else
				fctx->record[i].isdirty = false;

			/* Note if the buffer is valid, and has storage created */
			if ((bufHdr->flags & BM_VALID) && (bufHdr->flags & BM_TAG_VALID))
				fctx->record[i].isvalid = true;
			else
				fctx->record[i].isvalid = false;

			UnlockBufHdr(bufHdr);
		}

		/*
		 * And release locks.  We do this in reverse order for two reasons:
		 * (1) Anyone else who needs more than one of the locks will be trying
		 * to lock them in increasing order; we don't want to release the
		 * other process until it can get all the locks it needs. (2) This
		 * avoids O(N^2) behavior inside LWLockRelease.
		 */
		for (i = NUM_BUFFER_PARTITIONS; --i >= 0;)
			LWLockRelease(FirstBufMappingLock + i);
	}

	funcctx = SRF_PERCALL_SETUP();

	/* Get the saved state */
	fctx = funcctx->user_fctx;

	if (funcctx->call_cntr < funcctx->max_calls)
	{
		uint32		i = funcctx->call_cntr;
		Datum		values[NUM_BUFFERCACHE_PAGES_ELEM];
		bool		nulls[NUM_BUFFERCACHE_PAGES_ELEM];

		values[0] = Int32GetDatum(fctx->record[i].bufferid);
		nulls[0] = false;

		/*
		 * Set all fields except the bufferid to null if the buffer is unused
		 * or not valid.
		 */
		if (fctx->record[i].blocknum == InvalidBlockNumber ||
			fctx->record[i].isvalid == false)
		{
			nulls[1] = true;
			nulls[2] = true;
			nulls[3] = true;
			nulls[4] = true;
			nulls[5] = true;
			nulls[6] = true;
			nulls[7] = true;
		}
		else
		{
			values[1] = ObjectIdGetDatum(fctx->record[i].relfilenode);
			nulls[1] = false;
			values[2] = ObjectIdGetDatum(fctx->record[i].reltablespace);
			nulls[2] = false;
			values[3] = ObjectIdGetDatum(fctx->record[i].reldatabase);
			nulls[3] = false;
			values[4] = ObjectIdGetDatum(fctx->record[i].forknum);
			nulls[4] = false;
			values[5] = Int64GetDatum((int64) fctx->record[i].blocknum);
			nulls[5] = false;
			values[6] = BoolGetDatum(fctx->record[i].isdirty);
			nulls[6] = false;
			values[7] = Int16GetDatum(fctx->record[i].usagecount);
			nulls[7] = false;
		}

		/* Build and return the tuple. */
		tuple = heap_form_tuple(fctx->tupdesc, values, nulls);
		result = HeapTupleGetDatum(tuple);

		SRF_RETURN_NEXT(funcctx, result);
	}
	else
		SRF_RETURN_DONE(funcctx);
}

/*
 * sepgsql_schema_post_create
 *
 * This routine assigns a default security label on a newly defined
 * schema.
 */
void
sepgsql_schema_post_create(Oid namespaceId)
{
	Relation	rel;
	ScanKeyData skey;
	SysScanDesc sscan;
	HeapTuple	tuple;
	char	   *tcontext;
	char	   *ncontext;
	char		audit_name[NAMEDATALEN + 20];
	ObjectAddress object;
	Form_pg_namespace nspForm;

	/*
	 * Compute a default security label when we create a new schema object
	 * under the working database.
	 *
	 * XXX - uncoming version of libselinux supports to take object name to
	 * handle special treatment on default security label; such as special
	 * label on "pg_temp" schema.
	 */
	rel = heap_open(NamespaceRelationId, AccessShareLock);

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

	sscan = systable_beginscan(rel, NamespaceOidIndexId, true,
							   SnapshotSelf, 1, &skey);
	tuple = systable_getnext(sscan);
	if (!HeapTupleIsValid(tuple))
		elog(ERROR, "catalog lookup failed for namespace %u", namespaceId);

	nspForm = (Form_pg_namespace) GETSTRUCT(tuple);

	tcontext = sepgsql_get_label(DatabaseRelationId, MyDatabaseId, 0);
	ncontext = sepgsql_compute_create(sepgsql_get_client_label(),
									  tcontext,
									  SEPG_CLASS_DB_SCHEMA);

	/*
	 * check db_schema:{create}
	 */
	snprintf(audit_name, sizeof(audit_name),
			 "schema %s", NameStr(nspForm->nspname));
	sepgsql_avc_check_perms_label(ncontext,
								  SEPG_CLASS_DB_SCHEMA,
								  SEPG_DB_SCHEMA__CREATE,
								  audit_name,
								  true);
	systable_endscan(sscan);
	heap_close(rel, AccessShareLock);

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

	pfree(ncontext);
	pfree(tcontext);
}

static collation_cache_entry *
lookup_collation_cache(Oid collation, bool set_flags)
{
	collation_cache_entry *cache_entry;
	bool		found;

	Assert(OidIsValid(collation));
	Assert(collation != DEFAULT_COLLATION_OID);

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

		memset(&ctl, 0, sizeof(ctl));
		ctl.keysize = sizeof(Oid);
		ctl.entrysize = sizeof(collation_cache_entry);
		ctl.hash = oid_hash;
		collation_cache = hash_create("Collation cache", 100, &ctl,
									  HASH_ELEM | HASH_FUNCTION);
	}

	cache_entry = hash_search(collation_cache, &collation, HASH_ENTER, &found);
	if (!found)
	{
		/*
		 * Make sure cache entry is marked invalid, in case we fail before
		 * setting things.
		 */
		cache_entry->flags_valid = false;
		cache_entry->locale = 0;
	}

	if (set_flags && !cache_entry->flags_valid)
	{
		/* Attempt to set the flags */
		HeapTuple	tp;
		Form_pg_collation collform;
		const char *collcollate;
		const char *collctype;

		tp = SearchSysCache1(COLLOID, ObjectIdGetDatum(collation));
		if (!HeapTupleIsValid(tp))
			elog(ERROR, "cache lookup failed for collation %u", collation);
		collform = (Form_pg_collation) GETSTRUCT(tp);

		collcollate = NameStr(collform->collcollate);
		collctype = NameStr(collform->collctype);

		cache_entry->collate_is_c = ((strcmp(collcollate, "C") == 0) ||
									 (strcmp(collcollate, "POSIX") == 0));
		cache_entry->ctype_is_c = ((strcmp(collctype, "C") == 0) ||
								   (strcmp(collctype, "POSIX") == 0));

		cache_entry->flags_valid = true;

		ReleaseSysCache(tp);
	}

	return cache_entry;
}

/*
 * Create a locale_t from a collation OID.	Results are cached for the
 * lifetime of the backend.  Thus, do not free the result with freelocale().
 *
 * As a special optimization, the default/database collation returns 0.
 * Callers should then revert to the non-locale_t-enabled code path.
 * In fact, they shouldn't call this function at all when they are dealing
 * with the default locale.  That can save quite a bit in hotspots.
 * Also, callers should avoid calling this before going down a C/POSIX
 * fastpath, because such a fastpath should work even on platforms without
 * locale_t support in the C library.
 *
 * For simplicity, we always generate COLLATE + CTYPE even though we
 * might only need one of them.  Since this is called only once per session,
 * it shouldn't cost much.
 */
pg_locale_t
pg_newlocale_from_collation(Oid collid)
{
	collation_cache_entry *cache_entry;

	/* Callers must pass a valid OID */
	Assert(OidIsValid(collid));

	/* Return 0 for "default" collation, just in case caller forgets */
	if (collid == DEFAULT_COLLATION_OID)
		return (pg_locale_t) 0;

	cache_entry = lookup_collation_cache(collid, false);

	if (cache_entry->locale == 0)
	{
		/* We haven't computed this yet in this session, so do it */
#ifdef HAVE_LOCALE_T
		HeapTuple	tp;
		Form_pg_collation collform;
		const char *collcollate;
		const char *collctype;
		locale_t	result;

		tp = SearchSysCache1(COLLOID, ObjectIdGetDatum(collid));
		if (!HeapTupleIsValid(tp))
			elog(ERROR, "cache lookup failed for collation %u", collid);
		collform = (Form_pg_collation) GETSTRUCT(tp);

		collcollate = NameStr(collform->collcollate);
		collctype = NameStr(collform->collctype);

		if (strcmp(collcollate, collctype) == 0)
		{
			/* Normal case where they're the same */
#ifndef WIN32
			result = newlocale(LC_COLLATE_MASK | LC_CTYPE_MASK, collcollate,
							   NULL);
#else
			result = _create_locale(LC_ALL, collcollate);
#endif
			if (!result)
				report_newlocale_failure(collcollate);
		}
		else
		{
#ifndef WIN32
			/* We need two newlocale() steps */
			locale_t	loc1;

			loc1 = newlocale(LC_COLLATE_MASK, collcollate, NULL);
			if (!loc1)
				report_newlocale_failure(collcollate);
			result = newlocale(LC_CTYPE_MASK, collctype, loc1);
			if (!result)
				report_newlocale_failure(collctype);
#else

			/*
			 * XXX The _create_locale() API doesn't appear to support this.
			 * Could perhaps be worked around by changing pg_locale_t to
			 * contain two separate fields.
			 */
			ereport(ERROR,
					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
					 errmsg("collations with different collate and ctype values are not supported on this platform")));
#endif
		}

		cache_entry->locale = result;

		ReleaseSysCache(tp);
#else							/* not HAVE_LOCALE_T */

		/*
		 * For platforms that don't support locale_t, we can't do anything
		 * with non-default collations.
		 */
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
		errmsg("nondefault collations are not supported on this platform")));
#endif   /* not HAVE_LOCALE_T */
	}

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

/*
 * CREATE SCHEMA
 *
 * Note: caller should pass in location information for the whole
 * CREATE SCHEMA statement, which in turn we pass down as the location
 * of the component commands.  This comports with our general plan of
 * reporting location/len for the whole command even when executing
 * a subquery.
 */
Oid
CreateSchemaCommand(CreateSchemaStmt *stmt, const char *queryString,
					int stmt_location, int stmt_len)
{
	const char *schemaName = stmt->schemaname;
	Oid			namespaceId;
	OverrideSearchPath *overridePath;
	List	   *parsetree_list;
	ListCell   *parsetree_item;
	Oid			owner_uid;
	Oid			saved_uid;
	int			save_sec_context;
	AclResult	aclresult;
	ObjectAddress address;

	GetUserIdAndSecContext(&saved_uid, &save_sec_context);

	/*
	 * Who is supposed to own the new schema?
	 */
	if (stmt->authrole)
		owner_uid = get_rolespec_oid(stmt->authrole, false);
	else
		owner_uid = saved_uid;

	/* fill schema name with the user name if not specified */
	if (!schemaName)
	{
		HeapTuple	tuple;

		tuple = SearchSysCache1(AUTHOID, ObjectIdGetDatum(owner_uid));
		if (!HeapTupleIsValid(tuple))
			elog(ERROR, "cache lookup failed for role %u", owner_uid);
		schemaName =
			pstrdup(NameStr(((Form_pg_authid) GETSTRUCT(tuple))->rolname));
		ReleaseSysCache(tuple);
	}

	/*
	 * To create a schema, must have schema-create privilege on the current
	 * database and must be able to become the target role (this does not
	 * imply that the target role itself must have create-schema privilege).
	 * The latter provision guards against "giveaway" attacks.  Note that a
	 * superuser will always have both of these privileges a fortiori.
	 */
	aclresult = pg_database_aclcheck(MyDatabaseId, saved_uid, ACL_CREATE);
	if (aclresult != ACLCHECK_OK)
		aclcheck_error(aclresult, ACL_KIND_DATABASE,
					   get_database_name(MyDatabaseId));

	check_is_member_of_role(saved_uid, owner_uid);

	/* Additional check to protect reserved schema names */
	if (!allowSystemTableMods && IsReservedName(schemaName))
		ereport(ERROR,
				(errcode(ERRCODE_RESERVED_NAME),
				 errmsg("unacceptable schema name \"%s\"", schemaName),
		   errdetail("The prefix \"pg_\" is reserved for system schemas.")));

	/*
	 * If if_not_exists was given and the schema already exists, bail out.
	 * (Note: we needn't check this when not if_not_exists, because
	 * NamespaceCreate will complain anyway.)  We could do this before making
	 * the permissions checks, but since CREATE TABLE IF NOT EXISTS makes its
	 * creation-permission check first, we do likewise.
	 */
	if (stmt->if_not_exists &&
		SearchSysCacheExists1(NAMESPACENAME, PointerGetDatum(schemaName)))
	{
		ereport(NOTICE,
				(errcode(ERRCODE_DUPLICATE_SCHEMA),
				 errmsg("schema \"%s\" already exists, skipping",
						schemaName)));
		return InvalidOid;
	}

	/*
	 * If the requested authorization is different from the current user,
	 * temporarily set the current user so that the object(s) will be created
	 * with the correct ownership.
	 *
	 * (The setting will be restored at the end of this routine, or in case of
	 * error, transaction abort will clean things up.)
	 */
	if (saved_uid != owner_uid)
		SetUserIdAndSecContext(owner_uid,
							save_sec_context | SECURITY_LOCAL_USERID_CHANGE);

	/* Create the schema's namespace */
	namespaceId = NamespaceCreate(schemaName, owner_uid, false);

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

/*
 *	Handle CTIDs of views.
 *		CTID should be defined in the view and it must
 *		correspond to the CTID of a base relation.
 */
static Datum
currtid_for_view(Relation viewrel, ItemPointer tid)
{
	TupleDesc	att = RelationGetDescr(viewrel);
	RuleLock   *rulelock;
	RewriteRule *rewrite;
	int			i,
				natts = att->natts,
				tididx = -1;

	for (i = 0; i < natts; i++)
	{
		Form_pg_attribute attr = TupleDescAttr(att, i);

		if (strcmp(NameStr(attr->attname), "ctid") == 0)
		{
			if (attr->atttypid != TIDOID)
				elog(ERROR, "ctid isn't of type TID");
			tididx = i;
			break;
		}
	}
	if (tididx < 0)
		elog(ERROR, "currtid cannot handle views with no CTID");
	rulelock = viewrel->rd_rules;
	if (!rulelock)
		elog(ERROR, "the view has no rules");
	for (i = 0; i < rulelock->numLocks; i++)
	{
		rewrite = rulelock->rules[i];
		if (rewrite->event == CMD_SELECT)
		{
			Query	   *query;
			TargetEntry *tle;

			if (list_length(rewrite->actions) != 1)
				elog(ERROR, "only one select rule is allowed in views");
			query = (Query *) linitial(rewrite->actions);
			tle = get_tle_by_resno(query->targetList, tididx + 1);
			if (tle && tle->expr && IsA(tle->expr, Var))
			{
				Var		   *var = (Var *) tle->expr;
				RangeTblEntry *rte;

				if (!IS_SPECIAL_VARNO(var->varno) &&
					var->varattno == SelfItemPointerAttributeNumber)
				{
					rte = rt_fetch(var->varno, query->rtable);
					if (rte)
					{
						heap_close(viewrel, AccessShareLock);
						return DirectFunctionCall2(currtid_byreloid, ObjectIdGetDatum(rte->relid), PointerGetDatum(tid));
					}
				}
			}
			break;
		}
	}
	elog(ERROR, "currtid cannot handle this view");
	return (Datum) 0;
}

/*
 * shdepChangeDep
 *
 * Update shared dependency records to account for an updated referenced
 * object.  This is an internal workhorse for operations such as changing
 * an object's owner.
 *
 * There must be no more than one existing entry for the given dependent
 * object and dependency type!	So in practice this can only be used for
 * updating SHARED_DEPENDENCY_OWNER entries, which should have that property.
 *
 * If there is no previous entry, we assume it was referencing a PINned
 * object, so we create a new entry.  If the new referenced object is
 * PINned, we don't create an entry (and drop the old one, if any).
 *
 * sdepRel must be the pg_shdepend relation, already opened and suitably
 * locked.
 */
static void
shdepChangeDep(Relation sdepRel,
               Oid classid, Oid objid, int32 objsubid,
               Oid refclassid, Oid refobjid,
               SharedDependencyType deptype)
{
    Oid			dbid = classIdGetDbId(classid);
    HeapTuple	oldtup = NULL;
    HeapTuple	scantup;
    ScanKeyData key[4];
    SysScanDesc scan;

    /*
     * Make sure the new referenced object doesn't go away while we record the
     * dependency.
     */
    shdepLockAndCheckObject(refclassid, refobjid);

    /*
     * Look for a previous entry
     */
    ScanKeyInit(&key[0],
                Anum_pg_shdepend_dbid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(dbid));
    ScanKeyInit(&key[1],
                Anum_pg_shdepend_classid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(classid));
    ScanKeyInit(&key[2],
                Anum_pg_shdepend_objid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(objid));
    ScanKeyInit(&key[3],
                Anum_pg_shdepend_objsubid,
                BTEqualStrategyNumber, F_INT4EQ,
                Int32GetDatum(objsubid));

    scan = systable_beginscan(sdepRel, SharedDependDependerIndexId, true,
                              NULL, 4, key);

    while ((scantup = systable_getnext(scan)) != NULL)
    {
        /* Ignore if not of the target dependency type */
        if (((Form_pg_shdepend) GETSTRUCT(scantup))->deptype != deptype)
            continue;
        /* Caller screwed up if multiple matches */
        if (oldtup)
            elog(ERROR,
                 "multiple pg_shdepend entries for object %u/%u/%d deptype %c",
                 classid, objid, objsubid, deptype);
        oldtup = heap_copytuple(scantup);
    }

    systable_endscan(scan);

    if (isSharedObjectPinned(refclassid, refobjid, sdepRel))
    {
        /* No new entry needed, so just delete existing entry if any */
        if (oldtup)
            simple_heap_delete(sdepRel, &oldtup->t_self);
    }
    else if (oldtup)
    {
        /* Need to update existing entry */
        Form_pg_shdepend shForm = (Form_pg_shdepend) GETSTRUCT(oldtup);

        /* Since oldtup is a copy, we can just modify it in-memory */
        shForm->refclassid = refclassid;
        shForm->refobjid = refobjid;

        simple_heap_update(sdepRel, &oldtup->t_self, oldtup);

        /* keep indexes current */
        CatalogUpdateIndexes(sdepRel, oldtup);
    }
    else
    {
        /* Need to insert new entry */
        Datum		values[Natts_pg_shdepend];
        bool		nulls[Natts_pg_shdepend];

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

void
PLy_output_tuple_funcs(PLyTypeInfo *arg, TupleDesc desc)
{
	int			i;
	PLyExecutionContext *exec_ctx = PLy_current_execution_context();

	if (arg->is_rowtype == 0)
		elog(ERROR, "PLyTypeInfo struct is initialized for a Datum");
	arg->is_rowtype = 1;

	if (arg->out.r.natts != desc->natts)
	{
		if (arg->out.r.atts)
			PLy_free(arg->out.r.atts);
		arg->out.r.natts = desc->natts;
		arg->out.r.atts = PLy_malloc0(desc->natts * sizeof(PLyObToDatum));
	}

	Assert(OidIsValid(desc->tdtypeid));

	/*
	 * RECORDOID means we got called to create output functions for an
	 * anonymous record type
	 */
	if (desc->tdtypeid != RECORDOID)
	{
		HeapTuple	relTup;

		/* Get the pg_class tuple corresponding to the type of the output */
		arg->typ_relid = typeidTypeRelid(desc->tdtypeid);
		relTup = SearchSysCache1(RELOID, ObjectIdGetDatum(arg->typ_relid));
		if (!HeapTupleIsValid(relTup))
			elog(ERROR, "cache lookup failed for relation %u", arg->typ_relid);

		/* Remember XMIN and TID for later validation if cache is still OK */
		arg->typrel_xmin = HeapTupleHeaderGetRawXmin(relTup->t_data);
		arg->typrel_tid = relTup->t_self;

		ReleaseSysCache(relTup);
	}

	for (i = 0; i < desc->natts; i++)
	{
		HeapTuple	typeTup;

		if (desc->attrs[i]->attisdropped)
			continue;

		if (arg->out.r.atts[i].typoid == desc->attrs[i]->atttypid)
			continue;			/* already set up this entry */

		typeTup = SearchSysCache1(TYPEOID,
								  ObjectIdGetDatum(desc->attrs[i]->atttypid));
		if (!HeapTupleIsValid(typeTup))
			elog(ERROR, "cache lookup failed for type %u",
				 desc->attrs[i]->atttypid);

		PLy_output_datum_func2(&(arg->out.r.atts[i]), typeTup,
							   exec_ctx->curr_proc->langid,
							   exec_ctx->curr_proc->trftypes);

		ReleaseSysCache(typeTup);
	}
}

/*
 * checkSharedDependencies
 *
 * Check whether there are shared dependency entries for a given shared
 * object; return true if so.
 *
 * In addition, return a string containing a newline-separated list of object
 * descriptions that depend on the shared object, or NULL if none is found.
 * We actually return two such strings; the "detail" result is suitable for
 * returning to the client as an errdetail() string, and is limited in size.
 * The "detail_log" string is potentially much longer, and should be emitted
 * to the server log only.
 *
 * We can find three different kinds of dependencies: dependencies on objects
 * of the current database; dependencies on shared objects; and dependencies
 * on objects local to other databases.  We can (and do) provide descriptions
 * of the two former kinds of objects, but we can't do that for "remote"
 * objects, so we just provide a count of them.
 *
 * If we find a SHARED_DEPENDENCY_PIN entry, we can error out early.
 */
bool
checkSharedDependencies(Oid classId, Oid objectId,
                        char **detail_msg, char **detail_log_msg)
{
    Relation	sdepRel;
    ScanKeyData key[2];
    SysScanDesc scan;
    HeapTuple	tup;
    int			numReportedDeps = 0;
    int			numNotReportedDeps = 0;
    int			numNotReportedDbs = 0;
    List	   *remDeps = NIL;
    ListCell   *cell;
    ObjectAddress object;
    StringInfoData descs;
    StringInfoData alldescs;

    /*
     * We limit the number of dependencies reported to the client to
     * MAX_REPORTED_DEPS, since client software may not deal well with
     * enormous error strings.  The server log always gets a full report.
     */
#define MAX_REPORTED_DEPS 100

    initStringInfo(&descs);
    initStringInfo(&alldescs);

    sdepRel = heap_open(SharedDependRelationId, AccessShareLock);

    ScanKeyInit(&key[0],
                Anum_pg_shdepend_refclassid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(classId));
    ScanKeyInit(&key[1],
                Anum_pg_shdepend_refobjid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(objectId));

    scan = systable_beginscan(sdepRel, SharedDependReferenceIndexId, true,
                              NULL, 2, key);

    while (HeapTupleIsValid(tup = systable_getnext(scan)))
    {
        Form_pg_shdepend sdepForm = (Form_pg_shdepend) GETSTRUCT(tup);

        /* This case can be dispatched quickly */
        if (sdepForm->deptype == SHARED_DEPENDENCY_PIN)
        {
            object.classId = classId;
            object.objectId = objectId;
            object.objectSubId = 0;
            ereport(ERROR,
                    (errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
                     errmsg("cannot drop %s because it is required by the database system",
                            getObjectDescription(&object))));
        }

        object.classId = sdepForm->classid;
        object.objectId = sdepForm->objid;
        object.objectSubId = sdepForm->objsubid;

        /*
         * If it's a dependency local to this database or it's a shared
         * object, describe it.
         *
         * If it's a remote dependency, keep track of it so we can report the
         * number of them later.
         */
        if (sdepForm->dbid == MyDatabaseId)
        {
            if (numReportedDeps < MAX_REPORTED_DEPS)
            {
                numReportedDeps++;
                storeObjectDescription(&descs, LOCAL_OBJECT, &object,
                                       sdepForm->deptype, 0);
            }
            else
                numNotReportedDeps++;
            storeObjectDescription(&alldescs, LOCAL_OBJECT, &object,
//.........这里部分代码省略.........

/*
 * internal_get_result_type -- workhorse code implementing all the above
 *
 * funcid must always be supplied.  call_expr and rsinfo can be NULL if not
 * available.  We will return TYPEFUNC_RECORD, and store NULL into
 * *resultTupleDesc, if we cannot deduce the complete result rowtype from
 * the available information.
 */
static TypeFuncClass
internal_get_result_type(Oid funcid,
						 Node *call_expr,
						 ReturnSetInfo *rsinfo,
						 Oid *resultTypeId,
						 TupleDesc *resultTupleDesc)
{
	TypeFuncClass result;
	HeapTuple	tp;
	Form_pg_proc procform;
	Oid			rettype;
	TupleDesc	tupdesc;

	/* First fetch the function's pg_proc row to inspect its rettype */
	tp = SearchSysCache1(PROCOID, ObjectIdGetDatum(funcid));
	if (!HeapTupleIsValid(tp))
		elog(ERROR, "cache lookup failed for function %u", funcid);
	procform = (Form_pg_proc) GETSTRUCT(tp);

	rettype = procform->prorettype;

	/* Check for OUT parameters defining a RECORD result */
	tupdesc = build_function_result_tupdesc_t(tp);
	if (tupdesc)
	{
		/*
		 * It has OUT parameters, so it's basically like a regular composite
		 * type, except we have to be able to resolve any polymorphic OUT
		 * parameters.
		 */
		if (resultTypeId)
			*resultTypeId = rettype;

		if (resolve_polymorphic_tupdesc(tupdesc,
										&procform->proargtypes,
										call_expr))
		{
			if (tupdesc->tdtypeid == RECORDOID &&
				tupdesc->tdtypmod < 0)
				assign_record_type_typmod(tupdesc);
			if (resultTupleDesc)
				*resultTupleDesc = tupdesc;
			result = TYPEFUNC_COMPOSITE;
		}
		else
		{
			if (resultTupleDesc)
				*resultTupleDesc = NULL;
			result = TYPEFUNC_RECORD;
		}

		ReleaseSysCache(tp);

		return result;
	}

	/*
	 * If scalar polymorphic result, try to resolve it.
	 */
	if (IsPolymorphicType(rettype))
	{
		Oid			newrettype = exprType(call_expr);

		if (newrettype == InvalidOid)	/* this probably should not happen */
			ereport(ERROR,
					(errcode(ERRCODE_DATATYPE_MISMATCH),
					 errmsg("could not determine actual result type for function \"%s\" declared to return type %s",
							NameStr(procform->proname),
							format_type_be(rettype))));
		rettype = newrettype;
	}

	if (resultTypeId)
		*resultTypeId = rettype;
	if (resultTupleDesc)
		*resultTupleDesc = NULL;	/* default result */

	/* Classify the result type */
	result = get_type_func_class(rettype);
	switch (result)
	{
		case TYPEFUNC_COMPOSITE:
			if (resultTupleDesc)
				*resultTupleDesc = lookup_rowtype_tupdesc_copy(rettype, -1);
			/* Named composite types can't