自学教程：C++ BlessTupleDesc函数代码示例

Datumconvert_to_UTF8(PG_FUNCTION_ARGS){    // things we need to deal with constructing our composite type    TupleDesc   tupdesc;    Datum       values[3];    bool        nulls[3];    HeapTuple   tuple;    // for char_set_detect function returns    text    *encoding = NULL;    text    *lang = NULL;    int32_t confidence = 0;    UErrorCode status = U_ZERO_ERROR;    // output buffer for conversion to Unicode    UChar* uBuf = NULL;    int32_t uBuf_len = 0;    // output of this function    text *text_out;    bool converted = false;    bool dropped_bytes = false;    bool dropped_bytes_toU = false;    bool dropped_bytes_fromU = false;    // temporary buffer for converted string    char* converted_buf = NULL;    // input args    const text  *buffer = PG_GETARG_TEXT_P(0);    const bool  force   = PG_GETARG_BOOL(1);    // C string of text* buffer    const char* cbuffer = NULL;    int cbuffer_len = 0;    // Convert output values into a PostgreSQL composite type.    if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)        ereport(ERROR,            (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),              errmsg("function returning record called in context "                     "that cannot accept type record./n")));    // BlessTupleDesc for Datums    BlessTupleDesc(tupdesc);    // return if string to convert is NULL    if (NULL == buffer)    {        // return input string,        // converted to true,        // dropped_bytes to false        text_out = (text *) buffer;        converted = true;        dropped_bytes = false;    }    else    {        // extract string from text* to C string        cbuffer = text_to_cstring(buffer);        cbuffer_len = strlen(cbuffer);        // bail on zero-length strings        // return if cbuffer has zero length or contains a blank space        if ((0 == cbuffer_len) || (0 == strcmp("", cbuffer)))        {            text_out = (text *) buffer;            converted = true;            dropped_bytes = false;        }        else        {            // UTF8 output can be up to 6 bytes per input byte            // palloc0 allocates and zeros bytes in array            int32_t converted_buf_len = cbuffer_len * 6 * sizeof(char);            converted_buf = (char *) palloc0(converted_buf_len);            // int32_t converted_len = 0;            // detect encoding with ICU            status = detect_ICU(buffer, &encoding, &lang, &confidence);            ereport(DEBUG1,                (errcode(ERRCODE_SUCCESSFUL_COMPLETION),                 errmsg("ICU detection status: %d/n", status)));            ereport(DEBUG1,                (errcode(ERRCODE_SUCCESSFUL_COMPLETION),                 errmsg("Detected encoding: %s, language: %s, confidence: %d/n",                 text_to_cstring(encoding),                 text_to_cstring(lang),                         confidence)));            // return without attempting a conversion if UTF8 is detected            if (                (0 == strcmp("UTF-8", text_to_cstring(encoding))) ||                (0 == strcmp("utf-8", text_to_cstring(encoding))) ||                (0 == strcmp("UTF8", text_to_cstring(encoding)))  ||                (0 == strcmp("utf8", text_to_cstring(encoding)))//.........这里部分代码省略.........

/* * pgdatabasev - produce a view of pgdatabase to include transient state */Datumgp_pgdatabase__(PG_FUNCTION_ARGS){	FuncCallContext *funcctx;	Working_State *mystatus;	if (SRF_IS_FIRSTCALL())	{		TupleDesc	tupdesc;		MemoryContext oldcontext;		/* create a function context for cross-call persistence */		funcctx = SRF_FIRSTCALL_INIT();		/*		 * switch to memory context appropriate for multiple function		 * calls		 */		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);		/* build tupdesc for result tuples */		/* this had better match pg_prepared_xacts view in	system_views.sql */		tupdesc = CreateTemplateTupleDesc(5, false);		TupleDescInitEntry(tupdesc, (AttrNumber) 1, "dbid",						   INT2OID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 2, "isprimary",						   BOOLOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 3, "content",						   INT2OID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 4, "valid",						   BOOLOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 5, "definedprimary",						   BOOLOID, -1, 0);		funcctx->tuple_desc = BlessTupleDesc(tupdesc);		/*		 * Collect all the locking information that we will format and		 * send out as a result set.		 */		mystatus = (Working_State *) palloc(sizeof(Working_State));		funcctx->user_fctx = (void *) mystatus;		mystatus->master = GetMasterSegment();		mystatus->standby = GetStandbySegment();		mystatus->segments = GetSegmentList();		mystatus->idx = 0;		MemoryContextSwitchTo(oldcontext);	}	funcctx = SRF_PERCALL_SETUP();	mystatus = (Working_State *) funcctx->user_fctx;	while (mystatus->master || mystatus->standby || (mystatus->idx < list_length(mystatus->segments)))	{		Datum		values[6];		bool		nulls[6];		HeapTuple	tuple;		Datum		result;		Segment 	*current = NULL;		if (mystatus->master)		{			current = mystatus->master;			mystatus->master = NULL;		}		else if (mystatus->standby)		{			current = mystatus->standby;			mystatus->standby = NULL;		}		else		{			current = list_nth(mystatus->segments, mystatus->idx);			mystatus->idx++;		}		/*		 * Form tuple with appropriate data.		 */		MemSet(values, 0, sizeof(values));		MemSet(nulls, false, sizeof(nulls));		//values[0] = UInt16GetDatum(current->dbid);		values[1] = current->standby ? false : true;;		values[2] = UInt16GetDatum(current->segindex);		values[3] = BoolGetDatum(true);		values[4] = values[1];		tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);		result = HeapTupleGetDatum(tuple);		SRF_RETURN_NEXT(funcctx, result);	}	SRF_RETURN_DONE(funcctx);//.........这里部分代码省略.........

/* * stat a file */Datumpg_stat_file(PG_FUNCTION_ARGS){	text	   *filename_t = PG_GETARG_TEXT_P(0);	char	   *filename;	struct stat fst;	Datum		values[6];	bool		isnull[6];	HeapTuple	tuple;	TupleDesc	tupdesc;	if (!superuser())		ereport(ERROR,				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),				 (errmsg("must be superuser to get file information"))));	filename = convert_and_check_filename(filename_t);	if (stat(filename, &fst) < 0)		ereport(ERROR,				(errcode_for_file_access(),				 errmsg("could not stat file /"%s/": %m", filename)));	/*	 * This record type had better match the output parameters declared for me	 * in pg_proc.h.	 */	tupdesc = CreateTemplateTupleDesc(6, false);	TupleDescInitEntry(tupdesc, (AttrNumber) 1,					   "size", INT8OID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 2,					   "access", TIMESTAMPTZOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 3,					   "modification", TIMESTAMPTZOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 4,					   "change", TIMESTAMPTZOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 5,					   "creation", TIMESTAMPTZOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 6,					   "isdir", BOOLOID, -1, 0);	BlessTupleDesc(tupdesc);	memset(isnull, false, sizeof(isnull));	values[0] = Int64GetDatum((int64) fst.st_size);	values[1] = TimestampTzGetDatum(time_t_to_timestamptz(fst.st_atime));	values[2] = TimestampTzGetDatum(time_t_to_timestamptz(fst.st_mtime));	/* Unix has file status change time, while Win32 has creation time */#if !defined(WIN32) && !defined(__CYGWIN__)	values[3] = TimestampTzGetDatum(time_t_to_timestamptz(fst.st_ctime));	isnull[4] = true;#else	isnull[3] = true;	values[4] = TimestampTzGetDatum(time_t_to_timestamptz(fst.st_ctime));#endif	values[5] = BoolGetDatum(S_ISDIR(fst.st_mode));	tuple = heap_form_tuple(tupdesc, values, isnull);	pfree(filename);	PG_RETURN_DATUM(HeapTupleGetDatum(tuple));}

Datum extractGridData(PG_FUNCTION_ARGS){    FuncCallContext     *funcctx;     /* stuff done only on the first call of the function */     if (SRF_IS_FIRSTCALL())     {		TupleDesc tupdesc;        /* create a function context for cross-call persistence */        funcctx = SRF_FIRSTCALL_INIT();        /* Build a tuple descriptor for our result type */        if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)            ereport(ERROR,                    (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),                     errmsg("function returning record called in context "                            "that cannot accept type record")));        /* switch to memory context appropriate for multiple function calls */        MemoryContext oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);        funcctx->tuple_desc = BlessTupleDesc(tupdesc);        // Initalize geos        static int geosInitialized = 0;        if ( ! geosInitialized )        {        	initGEOS(logInfo, logError);        	geosInitialized = 1;        }        // Get the data to be returned        struct GridPointDataListIterator * points = getExtractGridDataReturnValues(fcinfo);        // Convert data into a set of Datums        funcctx->max_calls = points->list->count;        Datum * returnValues = palloc(sizeof(Datum) * funcctx->max_calls);        int i;        for ( i = 0; i < funcctx->max_calls; ++ i )        	returnValues[i] = getNextReturnTupleViaDatums(GridPointDataListIteratorNext(points), funcctx->tuple_desc);        funcctx->user_fctx = (void *) returnValues;        // Delete intermediate data//        GridPointDataListDelete(points->list);//        GridPointDataListIteratorDelete(points);        MemoryContextSwitchTo(oldcontext);     }    /* stuff done on every call of the function */    funcctx = SRF_PERCALL_SETUP();    if ( funcctx->call_cntr < funcctx->max_calls )    {    	Datum * ret = (Datum *) funcctx->user_fctx;    	Datum result = ret[funcctx->call_cntr];        SRF_RETURN_NEXT(funcctx, result);    }    else    {    	//GridPointDataListDelete(iterator->list);        SRF_RETURN_DONE(funcctx);    }}

/* * SQL function json_populate_recordset * * set fields in a set of records from the argument json, * which must be an array of objects. * * similar to json_populate_record, but the tuple-building code * is pushed down into the semantic action handlers so it's done * per object in the array. */Datumjson_populate_recordset(PG_FUNCTION_ARGS){	Oid			argtype = get_fn_expr_argtype(fcinfo->flinfo, 0);	text	   *json = PG_GETARG_TEXT_P(1);	bool		use_json_as_text = PG_GETARG_BOOL(2);	ReturnSetInfo *rsi;	MemoryContext old_cxt;	Oid			tupType;	int32		tupTypmod;	HeapTupleHeader rec;	TupleDesc	tupdesc;	RecordIOData *my_extra;	int			ncolumns;	JsonLexContext *lex;	JsonSemAction sem;	PopulateRecordsetState state;	if (!type_is_rowtype(argtype))		ereport(ERROR,				(errcode(ERRCODE_DATATYPE_MISMATCH),				 errmsg("first argument must be a rowtype")));	rsi = (ReturnSetInfo *) fcinfo->resultinfo;	if (!rsi || !IsA(rsi, ReturnSetInfo) ||		(rsi->allowedModes & SFRM_Materialize) == 0 ||		rsi->expectedDesc == NULL)		ereport(ERROR,				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),				 errmsg("set-valued function called in context that "						"cannot accept a set")));	rsi->returnMode = SFRM_Materialize;	/*	 * get the tupdesc from the result set info - it must be a record type	 * because we already checked that arg1 is a record type.	 */	(void) get_call_result_type(fcinfo, NULL, &tupdesc);	state = palloc0(sizeof(populateRecordsetState));	sem = palloc0(sizeof(jsonSemAction));	/* make these in a sufficiently long-lived memory context */	old_cxt = MemoryContextSwitchTo(rsi->econtext->ecxt_per_query_memory);	state->ret_tdesc = CreateTupleDescCopy(tupdesc);	BlessTupleDesc(state->ret_tdesc);	state->tuple_store =		tuplestore_begin_heap(rsi->allowedModes & SFRM_Materialize,							  false, work_mem);	MemoryContextSwitchTo(old_cxt);	/* if the json is null send back an empty set */	if (PG_ARGISNULL(1))		PG_RETURN_NULL();	if (PG_ARGISNULL(0))		rec = NULL;	else		rec = PG_GETARG_HEAPTUPLEHEADER(0);	tupType = tupdesc->tdtypeid;	tupTypmod = tupdesc->tdtypmod;	ncolumns = tupdesc->natts;	lex = makeJsonLexContext(json, true);	/*	 * We arrange to look up the needed I/O info just once per series of	 * calls, assuming the record type doesn't change underneath us.	 */	my_extra = (RecordIOData *) fcinfo->flinfo->fn_extra;	if (my_extra == NULL ||		my_extra->ncolumns != ncolumns)	{		fcinfo->flinfo->fn_extra =			MemoryContextAlloc(fcinfo->flinfo->fn_mcxt,							   sizeof(RecordIOData) - sizeof(ColumnIOData)							   + ncolumns * sizeof(ColumnIOData));		my_extra = (RecordIOData *) fcinfo->flinfo->fn_extra;		my_extra->record_type = InvalidOid;		my_extra->record_typmod = 0;	}	if (my_extra->record_type != tupType ||//.........这里部分代码省略.........

//.........这里部分代码省略.........	/*	 * tuple table initialization	 */	ExecInitResultTupleSlot(estate, &scanstate->ss.ps);	ExecInitScanTupleSlot(estate, &scanstate->ss);	/*	 * initialize child expressions	 */	scanstate->ss.ps.targetlist = (List *)		ExecInitExpr((Expr *) node->scan.plan.targetlist,					 (PlanState *) scanstate);	scanstate->ss.ps.qual = (List *)		ExecInitExpr((Expr *) node->scan.plan.qual,					 (PlanState *) scanstate);	/* Check if targetlist or qual contains a var node referencing the ctid column */	scanstate->cdb_want_ctid = contain_ctid_var_reference(&node->scan);    ItemPointerSet(&scanstate->cdb_fake_ctid, 0, 0);    ItemPointerSet(&scanstate->cdb_mark_ctid, 0, 0);	/*	 * get info about function	 */	rte = rt_fetch(node->scan.scanrelid, estate->es_range_table);	Assert(rte->rtekind == RTE_FUNCTION);	/*	 * Now determine if the function returns a simple or composite type, and	 * build an appropriate tupdesc.	 */	functypclass = get_expr_result_type(rte->funcexpr,										&funcrettype,										&tupdesc);	if (functypclass == TYPEFUNC_COMPOSITE)	{		/* Composite data type, e.g. a table's row type */		Assert(tupdesc);		/* Must copy it out of typcache for safety */		tupdesc = CreateTupleDescCopy(tupdesc);	}	else if (functypclass == TYPEFUNC_SCALAR)	{		/* Base data type, i.e. scalar */		char	   *attname = strVal(linitial(rte->eref->colnames));		tupdesc = CreateTemplateTupleDesc(1, false);		TupleDescInitEntry(tupdesc,						   (AttrNumber) 1,						   attname,						   funcrettype,						   -1,						   0);	}	else if (functypclass == TYPEFUNC_RECORD)	{		tupdesc = BuildDescFromLists(rte->eref->colnames,									 rte->funccoltypes,									 rte->funccoltypmods);	}	else	{		/* crummy error message, but parser should have caught this */		elog(ERROR, "function in FROM has unsupported return type");	}	/*	 * For RECORD results, make sure a typmod has been assigned.  (The	 * function should do this for itself, but let's cover things in case it	 * doesn't.)	 */	BlessTupleDesc(tupdesc);	scanstate->tupdesc = tupdesc;	ExecAssignScanType(&scanstate->ss, tupdesc);	/*	 * Other node-specific setup	 */	scanstate->tuplestorestate = NULL;	scanstate->funcexpr = ExecInitExpr((Expr *) rte->funcexpr,									   (PlanState *) scanstate);	/*	 * Initialize result tuple type and projection info.	 */	ExecAssignResultTypeFromTL(&scanstate->ss.ps);	ExecAssignScanProjectionInfo(&scanstate->ss);	initGpmonPktForFunctionScan((Plan *)node, &scanstate->ss.ps.gpmon_pkt, estate);		if (gp_resqueue_memory_policy != RESQUEUE_MEMORY_POLICY_NONE)	{		SPI_ReserveMemory(((Plan *)node)->operatorMemKB * 1024L);	}	return scanstate;}

/* * pg_lock_status - produce a view with one row per held or awaited lock mode */Datumpg_lock_status(PG_FUNCTION_ARGS){	FuncCallContext *funcctx;	PG_Lock_Status *mystatus;	LockData   *lockData;	PredicateLockData *predLockData;	if (SRF_IS_FIRSTCALL())	{		TupleDesc	tupdesc;		MemoryContext oldcontext;		/* create a function context for cross-call persistence */		funcctx = SRF_FIRSTCALL_INIT();		/*		 * switch to memory context appropriate for multiple function calls		 */		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);		/* build tupdesc for result tuples */		/* this had better match pg_locks view in system_views.sql */		tupdesc = CreateTemplateTupleDesc(NUM_LOCK_STATUS_COLUMNS, false);		TupleDescInitEntry(tupdesc, (AttrNumber) 1, "locktype",						   TEXTOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 2, "database",						   OIDOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 3, "relation",						   OIDOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 4, "page",						   INT4OID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 5, "tuple",						   INT2OID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 6, "virtualxid",						   TEXTOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 7, "transactionid",						   XIDOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 8, "classid",						   OIDOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 9, "objid",						   OIDOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 10, "objsubid",						   INT2OID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 11, "virtualtransaction",						   TEXTOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 12, "pid",						   INT4OID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 13, "mode",						   TEXTOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 14, "granted",						   BOOLOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 15, "fastpath",						   BOOLOID, -1, 0);		funcctx->tuple_desc = BlessTupleDesc(tupdesc);		/*		 * Collect all the locking information that we will format and send		 * out as a result set.		 */		mystatus = (PG_Lock_Status *) palloc(sizeof(PG_Lock_Status));		funcctx->user_fctx = (void *) mystatus;		mystatus->lockData = GetLockStatusData();		mystatus->currIdx = 0;		mystatus->predLockData = GetPredicateLockStatusData();		mystatus->predLockIdx = 0;		MemoryContextSwitchTo(oldcontext);	}	funcctx = SRF_PERCALL_SETUP();	mystatus = (PG_Lock_Status *) funcctx->user_fctx;	lockData = mystatus->lockData;	while (mystatus->currIdx < lockData->nelements)	{		bool		granted;		LOCKMODE	mode = 0;		const char *locktypename;		char		tnbuf[32];		Datum		values[NUM_LOCK_STATUS_COLUMNS];		bool		nulls[NUM_LOCK_STATUS_COLUMNS];		HeapTuple	tuple;		Datum		result;		LockInstanceData   *instance;		instance = &(lockData->locks[mystatus->currIdx]);		/*		 * Look to see if there are any held lock modes in this PROCLOCK. If		 * so, report, and destructively modify lockData so we don't report		 * again.		 */		granted = false;		if (instance->holdMask)//.........这里部分代码省略.........

Datum sampleNewTopics(PG_FUNCTION_ARGS){	int32 i, widx, wtopic, rtopic;	ArrayType * doc_arr = PG_GETARG_ARRAYTYPE_P(0);	ArrayType * topics_arr = PG_GETARG_ARRAYTYPE_P(1);	ArrayType * topic_d_arr = PG_GETARG_ARRAYTYPE_P(2);	ArrayType * global_count_arr = PG_GETARG_ARRAYTYPE_P(3);	ArrayType * topic_counts_arr = PG_GETARG_ARRAYTYPE_P(4);	int32 num_topics = PG_GETARG_INT32(5);	int32 dsize = PG_GETARG_INT32(6);	float8 alpha = PG_GETARG_FLOAT8(7);	float8 eta = PG_GETARG_FLOAT8(8);	if (ARR_NULLBITMAP(doc_arr) || ARR_NDIM(doc_arr) != 1 || 	    ARR_ELEMTYPE(doc_arr) != INT4OID ||	    ARR_NDIM(topics_arr) != 1 || ARR_ELEMTYPE(topics_arr) != INT4OID ||	    ARR_NULLBITMAP(topic_d_arr) || ARR_NDIM(topic_d_arr) != 1 || 	    ARR_ELEMTYPE(topic_d_arr) != INT4OID ||	    ARR_NULLBITMAP(global_count_arr) || ARR_NDIM(global_count_arr) != 1	    || ARR_ELEMTYPE(global_count_arr) != INT4OID ||	    ARR_NULLBITMAP(topic_counts_arr) || ARR_NDIM(topic_counts_arr) != 1	    || ARR_ELEMTYPE(topic_counts_arr) != INT4OID)		ereport(ERROR,		       (errcode(ERRCODE_INVALID_PARAMETER_VALUE),			errmsg("function /"%s/" called with invalid parameters",			       format_procedure(fcinfo->flinfo->fn_oid))));	// the document array	int32 * doc = (int32 *)ARR_DATA_PTR(doc_arr);	int32 len = ARR_DIMS(doc_arr)[0];	// array giving topic assignment to each word in document	int32 * topics = (int32 *)ARR_DATA_PTR(topics_arr);	// distribution of topics in document	int32 * topic_d = (int32 *)ARR_DATA_PTR(topic_d_arr);	// the word-topic count matrix	int32 * global_count = (int32 *)ARR_DATA_PTR(global_count_arr);	// total number of words assigned to each topic in the whole corpus	int32 * topic_counts = (int32 *)ARR_DATA_PTR(topic_counts_arr);	ArrayType * ret_topics_arr, * ret_topic_d_arr;	int32 * ret_topics, * ret_topic_d;	Datum * arr1 = palloc0(len * sizeof(Datum));	ret_topics_arr = construct_array(arr1,len,INT4OID,4,true,'i');	ret_topics = (int32 *)ARR_DATA_PTR(ret_topics_arr);	Datum * arr2 = palloc0(num_topics * sizeof(Datum));	ret_topic_d_arr = construct_array(arr2,num_topics,INT4OID,4,true,'i');	ret_topic_d = (int32 *)ARR_DATA_PTR(ret_topic_d_arr);	for (i=0; i!=len; i++) {		widx = doc[i];		if (widx < 1 || widx > dsize)		     ereport		      (ERROR,		       (errcode(ERRCODE_INVALID_PARAMETER_VALUE),			errmsg("function /"%s/" called with invalid parameters",			       format_procedure(fcinfo->flinfo->fn_oid))));		wtopic = topics[i];		rtopic = sampleTopic(num_topics,widx,wtopic,global_count,				     topic_d,topic_counts,alpha,eta);		// <sampleNewTopics error checking> 		ret_topics[i] = rtopic;		ret_topic_d[rtopic-1]++;	}	Datum values[2];	values[0] = PointerGetDatum(ret_topics_arr);	values[1] = PointerGetDatum(ret_topic_d_arr);	TupleDesc tuple;	if (get_call_result_type(fcinfo, NULL, &tuple) != TYPEFUNC_COMPOSITE)		ereport(ERROR,			(errcode( ERRCODE_FEATURE_NOT_SUPPORTED ),			 errmsg( "function returning record called in context "				 "that cannot accept type record" )));	tuple = BlessTupleDesc(tuple);	bool * isnulls = palloc0(2 * sizeof(bool));	HeapTuple ret = heap_form_tuple(tuple, values, isnulls);	if (isnulls[0] || isnulls[1])		ereport(ERROR,			(errcode(ERRCODE_INVALID_PARAMETER_VALUE),			 errmsg("function /"%s/" produced null results",				format_procedure(fcinfo->flinfo->fn_oid),i)));	PG_RETURN_DATUM(HeapTupleGetDatum(ret));}

Datumpg_control_init(PG_FUNCTION_ARGS){	Datum		values[12];	bool		nulls[12];	TupleDesc	tupdesc;	HeapTuple	htup;	ControlFileData *ControlFile;	bool		crc_ok;	/*	 * Construct a tuple descriptor for the result row.  This must match this	 * function's pg_proc entry!	 */	tupdesc = CreateTemplateTupleDesc(12);	TupleDescInitEntry(tupdesc, (AttrNumber) 1, "max_data_alignment",					   INT4OID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 2, "database_block_size",					   INT4OID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 3, "blocks_per_segment",					   INT4OID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 4, "wal_block_size",					   INT4OID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 5, "bytes_per_wal_segment",					   INT4OID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 6, "max_identifier_length",					   INT4OID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 7, "max_index_columns",					   INT4OID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 8, "max_toast_chunk_size",					   INT4OID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 9, "large_object_chunk_size",					   INT4OID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 10, "float4_pass_by_value",					   BOOLOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 11, "float8_pass_by_value",					   BOOLOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 12, "data_page_checksum_version",					   INT4OID, -1, 0);	tupdesc = BlessTupleDesc(tupdesc);	/* read the control file */	ControlFile = get_controlfile(DataDir, NULL, &crc_ok);	if (!crc_ok)		ereport(ERROR,				(errmsg("calculated CRC checksum does not match value stored in file")));	values[0] = Int32GetDatum(ControlFile->maxAlign);	nulls[0] = false;	values[1] = Int32GetDatum(ControlFile->blcksz);	nulls[1] = false;	values[2] = Int32GetDatum(ControlFile->relseg_size);	nulls[2] = false;	values[3] = Int32GetDatum(ControlFile->xlog_blcksz);	nulls[3] = false;	values[4] = Int32GetDatum(ControlFile->xlog_seg_size);	nulls[4] = false;	values[5] = Int32GetDatum(ControlFile->nameDataLen);	nulls[5] = false;	values[6] = Int32GetDatum(ControlFile->indexMaxKeys);	nulls[6] = false;	values[7] = Int32GetDatum(ControlFile->toast_max_chunk_size);	nulls[7] = false;	values[8] = Int32GetDatum(ControlFile->loblksize);	nulls[8] = false;	values[9] = BoolGetDatum(ControlFile->float4ByVal);	nulls[9] = false;	values[10] = BoolGetDatum(ControlFile->float8ByVal);	nulls[10] = false;	values[11] = Int32GetDatum(ControlFile->data_checksum_version);	nulls[11] = false;	htup = heap_form_tuple(tupdesc, values, nulls);	PG_RETURN_DATUM(HeapTupleGetDatum(htup));}

Datumpg_control_checkpoint(PG_FUNCTION_ARGS){	Datum		values[19];	bool		nulls[19];	TupleDesc	tupdesc;	HeapTuple	htup;	ControlFileData *ControlFile;	XLogSegNo	segno;	char		xlogfilename[MAXFNAMELEN];	bool		crc_ok;	/*	 * Construct a tuple descriptor for the result row.  This must match this	 * function's pg_proc entry!	 */	tupdesc = CreateTemplateTupleDesc(18);	TupleDescInitEntry(tupdesc, (AttrNumber) 1, "checkpoint_lsn",					   LSNOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 2, "redo_lsn",					   LSNOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 3, "redo_wal_file",					   TEXTOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 4, "timeline_id",					   INT4OID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 5, "prev_timeline_id",					   INT4OID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 6, "full_page_writes",					   BOOLOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 7, "next_xid",					   TEXTOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 8, "next_oid",					   OIDOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 9, "next_multixact_id",					   XIDOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 10, "next_multi_offset",					   XIDOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 11, "oldest_xid",					   XIDOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 12, "oldest_xid_dbid",					   OIDOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 13, "oldest_active_xid",					   XIDOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 14, "oldest_multi_xid",					   XIDOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 15, "oldest_multi_dbid",					   OIDOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 16, "oldest_commit_ts_xid",					   XIDOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 17, "newest_commit_ts_xid",					   XIDOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 18, "checkpoint_time",					   TIMESTAMPTZOID, -1, 0);	tupdesc = BlessTupleDesc(tupdesc);	/* Read the control file. */	ControlFile = get_controlfile(DataDir, NULL, &crc_ok);	if (!crc_ok)		ereport(ERROR,				(errmsg("calculated CRC checksum does not match value stored in file")));	/*	 * Calculate name of the WAL file containing the latest checkpoint's REDO	 * start point.	 */	XLByteToSeg(ControlFile->checkPointCopy.redo, segno, wal_segment_size);	XLogFileName(xlogfilename, ControlFile->checkPointCopy.ThisTimeLineID,				 segno, wal_segment_size);	/* Populate the values and null arrays */	values[0] = LSNGetDatum(ControlFile->checkPoint);	nulls[0] = false;	values[1] = LSNGetDatum(ControlFile->checkPointCopy.redo);	nulls[1] = false;	values[2] = CStringGetTextDatum(xlogfilename);	nulls[2] = false;	values[3] = Int32GetDatum(ControlFile->checkPointCopy.ThisTimeLineID);	nulls[3] = false;	values[4] = Int32GetDatum(ControlFile->checkPointCopy.PrevTimeLineID);	nulls[4] = false;	values[5] = BoolGetDatum(ControlFile->checkPointCopy.fullPageWrites);	nulls[5] = false;	values[6] = CStringGetTextDatum(psprintf("%u:%u",											 ControlFile->checkPointCopy.nextXidEpoch,											 ControlFile->checkPointCopy.nextXid));	nulls[6] = false;	values[7] = ObjectIdGetDatum(ControlFile->checkPointCopy.nextOid);	nulls[7] = false;	values[8] = TransactionIdGetDatum(ControlFile->checkPointCopy.nextMulti);	nulls[8] = false;	values[9] = TransactionIdGetDatum(ControlFile->checkPointCopy.nextMultiOffset);//.........这里部分代码省略.........

Datum text_unpivot(PG_FUNCTION_ARGS){	FuncCallContext      *funcctx;	unpivot_fctx         *fctx;	MemoryContext         oldcontext;	Datum                 d[2];	bool                  isna[2];	/* stuff done only on the first call of the function */	if (SRF_IS_FIRSTCALL())	{		TupleDesc  tupdesc;		ArrayType *labels;		ArrayType *data;		Oid        eltype1;		Oid        eltype2;		/* see if we were given an explicit step size */		if (PG_NARGS() != 2)			ereport(ERROR,					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),					 errmsg("number of parameters != 2")));		/* 		 * Type check inputs:		 *    0: label text[]		 *    1: value anyarray		 */		eltype1 = get_fn_expr_argtype(fcinfo->flinfo, 0);		eltype2 = get_fn_expr_argtype(fcinfo->flinfo, 1);		if (!OidIsValid(eltype1))			elog(ERROR, "could not determine data type of input 'label'");		if (!OidIsValid(eltype2))			elog(ERROR, "could not determine data type of input 'value'");		/* Strict function, return null on null input */		if (PG_ARGISNULL(0) || PG_ARGISNULL(1))			PG_RETURN_NULL();				/* create a function context for cross-call persistence */		funcctx = SRF_FIRSTCALL_INIT();		/* switch to memory context appropriate for multiple function calls */		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);        /* Build a tuple descriptor for our result type */        if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)            ereport(ERROR,                    (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),                     errmsg("function returning record called in context "                            "that cannot accept type record")));		funcctx->tuple_desc = BlessTupleDesc(tupdesc);		/* allocate memory for user context */		fctx = (unpivot_fctx *) palloc(sizeof(unpivot_fctx));		/* Use fctx to keep state from call to call */		labels = PG_GETARG_ARRAYTYPE_P(0);		data   = PG_GETARG_ARRAYTYPE_P(1);		array_loop(labels, ARR_LBOUND(labels)[0], &fctx->label_iter);		array_loop(data,   ARR_LBOUND(labels)[0], &fctx->data_iter);				funcctx->user_fctx = fctx;		MemoryContextSwitchTo(oldcontext);	}	/* stuff done on every call of the function */	funcctx = SRF_PERCALL_SETUP();	/* get the saved state and use current as the result for this iteration */	fctx = (unpivot_fctx*) funcctx->user_fctx;	if (array_next(&fctx->label_iter, &d[0], &isna[0]))	{		HeapTuple tuple;		array_next(&fctx->data_iter, &d[1], &isna[1]);		tuple = heap_form_tuple(funcctx->tuple_desc, d, isna);		SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(tuple));			}	else	{		SRF_RETURN_DONE(funcctx);	}}

/* * Similar function exec_get_datum_type is in 9nth line */static Oidexec_get_datum_type(PLpgSQL_execstate *estate,				    PLpgSQL_datum *datum){	Oid typoid = InvalidOid;	switch (datum->dtype)	{		case PLPGSQL_DTYPE_VAR:			typoid = ((PLpgSQL_var *) datum)->datatype->typoid;			break;		case PLPGSQL_DTYPE_ROW:			typoid = ((PLpgSQL_row *) datum)->rowtupdesc->tdtypeid;			break;		case PLPGSQL_DTYPE_REC:			{				PLpgSQL_rec *rec = (PLpgSQL_rec *) datum;				if (!HeapTupleIsValid(rec->tup))					ereport(ERROR,						  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),						   errmsg("record /"%s/" is not assigned yet",								  rec->refname),						   errdetail("The tuple structure of a not-yet-assigned record is indeterminate.")));				Assert(rec->tupdesc != NULL);				/* Make sure we have a valid type/typmod setting */				BlessTupleDesc(rec->tupdesc);				typoid = rec->tupdesc->tdtypeid;			}			break;		case PLPGSQL_DTYPE_RECFIELD:			{				PLpgSQL_recfield *recfield = (PLpgSQL_recfield *) datum;				PLpgSQL_rec *rec;				int			fno;				rec = (PLpgSQL_rec *) (estate->datums[recfield->recparentno]);				if (!HeapTupleIsValid(rec->tup))					ereport(ERROR,						  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),						   errmsg("record /"%s/" is not assigned yet",								  rec->refname),						   errdetail("The tuple structure of a not-yet-assigned record is indeterminate.")));				fno = SPI_fnumber(rec->tupdesc, recfield->fieldname);				if (fno == SPI_ERROR_NOATTRIBUTE)					ereport(ERROR,							(errcode(ERRCODE_UNDEFINED_COLUMN),							 errmsg("record /"%s/" has no field /"%s/"",									rec->refname, recfield->fieldname)));				typoid = SPI_gettypeid(rec->tupdesc, fno);			}			break;		case PLPGSQL_DTYPE_TRIGARG:			typoid = TEXTOID;			break;	}	return typoid;}

Datumpg_stat_get_archiver(PG_FUNCTION_ARGS){	TupleDesc	tupdesc;	Datum		values[7];	bool		nulls[7];	PgStat_ArchiverStats *archiver_stats;	/* Initialise values and NULL flags arrays */	MemSet(values, 0, sizeof(values));	MemSet(nulls, 0, sizeof(nulls));	/* Initialise attributes information in the tuple descriptor */	tupdesc = CreateTemplateTupleDesc(7, false);	TupleDescInitEntry(tupdesc, (AttrNumber) 1, "archived_count",					   INT8OID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 2, "last_archived_wal",					   TEXTOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 3, "last_archived_time",					   TIMESTAMPTZOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 4, "failed_count",					   INT8OID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 5, "last_failed_wal",					   TEXTOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 6, "last_failed_time",					   TIMESTAMPTZOID, -1, 0);	TupleDescInitEntry(tupdesc, (AttrNumber) 7, "stats_reset",					   TIMESTAMPTZOID, -1, 0);	BlessTupleDesc(tupdesc);	/* Get statistics about the archiver process */	archiver_stats = pgstat_fetch_stat_archiver();	/* Fill values and NULLs */	values[0] = Int64GetDatum(archiver_stats->archived_count);	if (*(archiver_stats->last_archived_wal) == '/0')		nulls[1] = true;	else		values[1] = CStringGetTextDatum(archiver_stats->last_archived_wal);	if (archiver_stats->last_archived_timestamp == 0)		nulls[2] = true;	else		values[2] = TimestampTzGetDatum(archiver_stats->last_archived_timestamp);	values[3] = Int64GetDatum(archiver_stats->failed_count);	if (*(archiver_stats->last_failed_wal) == '/0')		nulls[4] = true;	else		values[4] = CStringGetTextDatum(archiver_stats->last_failed_wal);	if (archiver_stats->last_failed_timestamp == 0)		nulls[5] = true;	else		values[5] = TimestampTzGetDatum(archiver_stats->last_failed_timestamp);	if (archiver_stats->stat_reset_timestamp == 0)		nulls[6] = true;	else		values[6] = TimestampTzGetDatum(archiver_stats->stat_reset_timestamp);	/* Returns the record as Datum */	PG_RETURN_DATUM(HeapTupleGetDatum(								   heap_form_tuple(tupdesc, values, nulls)));}

/* * master_get_active_worker_nodes returns a set of active worker host names and * port numbers in deterministic order. Currently we assume that all worker * nodes in pg_worker_list.conf are active. */Datummaster_get_active_worker_nodes(PG_FUNCTION_ARGS){	FuncCallContext *functionContext = NULL;	uint32 workerNodeIndex = 0;	uint32 workerNodeCount = 0;	if (SRF_IS_FIRSTCALL())	{		MemoryContext oldContext = NULL;		List *workerNodeList = NIL;		uint32 workerNodeCount = 0;		TupleDesc tupleDescriptor = NULL;		bool hasOid = false;		/* create a function context for cross-call persistence */		functionContext = SRF_FIRSTCALL_INIT();		/* switch to memory context appropriate for multiple function calls */		oldContext = MemoryContextSwitchTo(functionContext->multi_call_memory_ctx);		workerNodeList = WorkerNodeList();		workerNodeCount = (uint32) list_length(workerNodeList);		functionContext->user_fctx = workerNodeList;		functionContext->max_calls = workerNodeCount;		/*		 * This tuple descriptor must match the output parameters declared for		 * the function in pg_proc.		 */		tupleDescriptor = CreateTemplateTupleDesc(WORKER_NODE_FIELDS, hasOid);		TupleDescInitEntry(tupleDescriptor, (AttrNumber) 1, "node_name",						   TEXTOID, -1, 0);		TupleDescInitEntry(tupleDescriptor, (AttrNumber) 2, "node_port",						   INT8OID, -1, 0);		functionContext->tuple_desc = BlessTupleDesc(tupleDescriptor);		MemoryContextSwitchTo(oldContext);	}	functionContext = SRF_PERCALL_SETUP();	workerNodeIndex = functionContext->call_cntr;	workerNodeCount = functionContext->max_calls;	if (workerNodeIndex < workerNodeCount)	{		List *workerNodeList = functionContext->user_fctx;		WorkerNode *workerNode = list_nth(workerNodeList, workerNodeIndex);		Datum workerNodeDatum = WorkerNodeGetDatum(workerNode,												   functionContext->tuple_desc);		SRF_RETURN_NEXT(functionContext, workerNodeDatum);	}	else	{		SRF_RETURN_DONE(functionContext);	}}

//.........这里部分代码省略.........  int                  max_calls;  TupleDesc            tuple_desc;  path_element_t      *path = 0;  /* stuff done only on the first call of the function */  if (SRF_IS_FIRSTCALL()) {    MemoryContext   oldcontext;    int path_count = 0;    int ret;        // XXX profiling messages are not thread safe    profstart(prof_total);    profstart(prof_extract);        /* create a function context for cross-call persistence */    funcctx = SRF_FIRSTCALL_INIT();        /* switch to memory context appropriate for multiple function calls */    oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);        ret = compute_driving_distance(text2char(PG_GETARG_TEXT_P(0)), // sql                                PG_GETARG_INT32(1),   // source vertex                                PG_GETARG_FLOAT8(2),  // distance or time                                PG_GETARG_BOOL(3),                                PG_GETARG_BOOL(4), &path, &path_count);    if (ret < 0) {        elog(ERROR, "Error computing path");    }     #ifdef DEBUG    DBG("Ret is %i", ret);    int i;    for (i = 0; i < path_count; i++) {        DBG("Step %i vertex_id  %i ", i, path[i].vertex_id);        DBG("        edge_id    %i ", path[i].edge_id);        DBG("        cost       %f ", path[i].cost);    }#endif    /* total number of tuples to be returned */    funcctx->max_calls = path_count;    funcctx->user_fctx = path;    funcctx->tuple_desc = BlessTupleDesc(                             RelationNameGetTupleDesc("pgr_costResult"));        MemoryContextSwitchTo(oldcontext);  }    /* stuff done on every call of the function */  funcctx = SRF_PERCALL_SETUP();  call_cntr = funcctx->call_cntr;  max_calls = funcctx->max_calls;  tuple_desc = funcctx->tuple_desc;  path = (path_element_t*) funcctx->user_fctx;    if (call_cntr < max_calls) {   /* do when there is more left to send */    HeapTuple    tuple;    Datum        result;    Datum *values;    char* nulls;        values = palloc(4 * sizeof(Datum));    nulls = palloc(4 * sizeof(char));    values[0] = Int32GetDatum(call_cntr);    nulls[0] = ' ';    values[1] = Int32GetDatum(path[call_cntr].vertex_id);    nulls[1] = ' ';    values[2] = Int32GetDatum(path[call_cntr].edge_id);    nulls[2] = ' ';    values[3] = Float8GetDatum(path[call_cntr].cost);    nulls[3] = ' ';          tuple = heap_formtuple(tuple_desc, values, nulls);        /* make the tuple into a datum */    result = HeapTupleGetDatum(tuple);        /* clean up (this is not really necessary) */    pfree(values);    pfree(nulls);    SRF_RETURN_NEXT(funcctx, result);  }  else {    /* do when there is no more left */    if (path) free(path);    profstop("store", prof_store);    profstop("total", prof_total);#ifdef PROFILE    elog(NOTICE, "_________");#endif    DBG("Returning value");    SRF_RETURN_DONE(funcctx);  }}

/* ---------------------------------------------------------------- *		ExecInitFunctionScan * ---------------------------------------------------------------- */FunctionScanState *ExecInitFunctionScan(FunctionScan *node, EState *estate, int eflags){	FunctionScanState *scanstate;	Oid			funcrettype;	TypeFuncClass functypclass;	TupleDesc	tupdesc = NULL;	/* check for unsupported flags */	Assert(!(eflags & EXEC_FLAG_MARK));	/*	 * FunctionScan should not have any children.	 */	Assert(outerPlan(node) == NULL);	Assert(innerPlan(node) == NULL);	/*	 * create new ScanState for node	 */	scanstate = makeNode(FunctionScanState);	scanstate->ss.ps.plan = (Plan *) node;	scanstate->ss.ps.state = estate;	scanstate->eflags = eflags;	/*	 * Miscellaneous initialization	 *	 * create expression context for node	 */	ExecAssignExprContext(estate, &scanstate->ss.ps);	/*	 * tuple table initialization	 */	ExecInitResultTupleSlot(estate, &scanstate->ss.ps);	ExecInitScanTupleSlot(estate, &scanstate->ss);	/*	 * initialize child expressions	 */	scanstate->ss.ps.targetlist = (List *)		ExecInitExpr((Expr *) node->scan.plan.targetlist,					 (PlanState *) scanstate);	scanstate->ss.ps.qual = (List *)		ExecInitExpr((Expr *) node->scan.plan.qual,					 (PlanState *) scanstate);	/*	 * Now determine if the function returns a simple or composite type, and	 * build an appropriate tupdesc.	 */	functypclass = get_expr_result_type(node->funcexpr,										&funcrettype,										&tupdesc);	if (functypclass == TYPEFUNC_COMPOSITE)	{		/* Composite data type, e.g. a table's row type */		Assert(tupdesc);		/* Must copy it out of typcache for safety */		tupdesc = CreateTupleDescCopy(tupdesc);	}	else if (functypclass == TYPEFUNC_SCALAR)	{		/* Base data type, i.e. scalar */		char	   *attname = strVal(linitial(node->funccolnames));		tupdesc = CreateTemplateTupleDesc(1, false);		TupleDescInitEntry(tupdesc,						   (AttrNumber) 1,						   attname,						   funcrettype,						   -1,						   0);	}	else if (functypclass == TYPEFUNC_RECORD)	{		tupdesc = BuildDescFromLists(node->funccolnames,									 node->funccoltypes,									 node->funccoltypmods);	}	else	{		/* crummy error message, but parser should have caught this */		elog(ERROR, "function in FROM has unsupported return type");	}	/*	 * For RECORD results, make sure a typmod has been assigned.  (The	 * function should do this for itself, but let's cover things in case it	 * doesn't.)	 */	BlessTupleDesc(tupdesc);	scanstate->tupdesc = tupdesc;//.........这里部分代码省略.........

Datumreadindex(PG_FUNCTION_ARGS){	FuncCallContext	   *funcctx;	readindexinfo	   *info;	Relation	irel = NULL;	Relation	hrel = NULL;	MIRROREDLOCK_BUFMGR_DECLARE;	if (SRF_IS_FIRSTCALL())	{		Oid		irelid = PG_GETARG_OID(0);		TupleDesc	tupdesc;		MemoryContext oldcontext;		AttrNumber		outattnum;		TupleDesc	itupdesc;		int			i;		AttrNumber	attno;		irel = index_open(irelid, AccessShareLock);		itupdesc = RelationGetDescr(irel);		outattnum = FIXED_COLUMN + itupdesc->natts;		funcctx = SRF_FIRSTCALL_INIT();		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);		tupdesc = CreateTemplateTupleDesc(outattnum, false);		attno = 1;		TupleDescInitEntry(tupdesc, attno++, "ictid", TIDOID, -1, 0);		TupleDescInitEntry(tupdesc, attno++, "hctid", TIDOID, -1, 0);		TupleDescInitEntry(tupdesc, attno++, "aotid", TEXTOID, -1, 0);		TupleDescInitEntry(tupdesc, attno++, "istatus", TEXTOID, -1, 0);		TupleDescInitEntry(tupdesc, attno++, "hstatus", TEXTOID, -1, 0);		for (i = 0; i < itupdesc->natts; i++)		{			Form_pg_attribute attr = itupdesc->attrs[i];			TupleDescInitEntry(tupdesc, attno++, NameStr(attr->attname), attr->atttypid, attr->atttypmod, 0);		}		funcctx->tuple_desc = BlessTupleDesc(tupdesc);		info = (readindexinfo *) palloc(sizeof(readindexinfo));		funcctx->user_fctx = (void *) info;		info->outattnum = outattnum;		info->ireloid = irelid;		hrel = relation_open(irel->rd_index->indrelid, AccessShareLock);		if (hrel->rd_rel != NULL &&			(hrel->rd_rel->relstorage == 'a' ||			 hrel->rd_rel->relstorage == 'c'))		{			relation_close(hrel, AccessShareLock);			hrel = NULL;			info->hreloid = InvalidOid;		}		else			info->hreloid = irel->rd_index->indrelid;		info->num_pages = RelationGetNumberOfBlocks(irel);		info->blkno = BTREE_METAPAGE + 1;		info->page = NULL;		MemoryContextSwitchTo(oldcontext);	}	funcctx = SRF_PERCALL_SETUP();	info = (readindexinfo *) funcctx->user_fctx;	/*	 * Open the relations (on first call, we did that above already).	 * We unfortunately have to look up the relcache entry on every call,	 * because if we store it in the cross-call context, we won't get a	 * chance to release it if the function isn't run to completion,	 * e.g. because of a LIMIT clause. We only lock the relation on the	 * first call, and keep the lock until completion, however.	 */	if (!irel)		irel = index_open(info->ireloid, NoLock);	if (!hrel && info->hreloid != InvalidOid)		hrel = heap_open(info->hreloid, NoLock);	while (info->blkno < info->num_pages)	{		Datum		values[255];		bool		nulls[255];		ItemPointerData		itid;		HeapTuple	tuple;		Datum		result;		if (info->page == NULL)		{			Buffer		buf;			/*			 * Make copy of the page, because we cannot hold a buffer pin			 * across calls (we wouldn't have a chance to release it, if the			 * function isn't run to completion.)			 */			info->page = palloc(BLCKSZ);//.........这里部分代码省略.........

Datumpg_stat_get_activity(PG_FUNCTION_ARGS){	FuncCallContext *funcctx;	if (SRF_IS_FIRSTCALL())	{		MemoryContext oldcontext;		TupleDesc	tupdesc;		funcctx = SRF_FIRSTCALL_INIT();		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);		tupdesc = CreateTemplateTupleDesc(14, false);		TupleDescInitEntry(tupdesc, (AttrNumber) 1, "datid",						   OIDOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 2, "pid",						   INT4OID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 3, "usesysid",						   OIDOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 4, "application_name",						   TEXTOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 5, "state",						   TEXTOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 6, "query",						   TEXTOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 7, "waiting",						   BOOLOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 8, "act_start",						   TIMESTAMPTZOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 9, "query_start",						   TIMESTAMPTZOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 10, "backend_start",						   TIMESTAMPTZOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 11, "state_change",						   TIMESTAMPTZOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 12, "client_addr",						   INETOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 13, "client_hostname",						   TEXTOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 14, "client_port",						   INT4OID, -1, 0);		funcctx->tuple_desc = BlessTupleDesc(tupdesc);		funcctx->user_fctx = palloc0(sizeof(int));		if (PG_ARGISNULL(0))		{			/* Get all backends */			funcctx->max_calls = pgstat_fetch_stat_numbackends();		}		else		{			/*			 * Get one backend - locate by pid.			 *			 * We lookup the backend early, so we can return zero rows if it			 * doesn't exist, instead of returning a single row full of NULLs.			 */			int			pid = PG_GETARG_INT32(0);			int			i;			int			n = pgstat_fetch_stat_numbackends();			for (i = 1; i <= n; i++)			{				PgBackendStatus *be = pgstat_fetch_stat_beentry(i);				if (be)				{					if (be->st_procpid == pid)					{						*(int *) (funcctx->user_fctx) = i;						break;					}				}			}			if (*(int *) (funcctx->user_fctx) == 0)				/* Pid not found, return zero rows */				funcctx->max_calls = 0;			else				funcctx->max_calls = 1;		}		MemoryContextSwitchTo(oldcontext);	}	/* stuff done on every call of the function */	funcctx = SRF_PERCALL_SETUP();	if (funcctx->call_cntr < funcctx->max_calls)	{		/* for each row */		Datum		values[14];		bool		nulls[14];		HeapTuple	tuple;		PgBackendStatus *beentry;		SockAddr	zero_clientaddr;//.........这里部分代码省略.........

Datum summarize_variant( PG_FUNCTION_ARGS ) {	if( PG_ARGISNULL(0) ) {		ereport( ERROR, (errmsg("summarize_variant: array of values must be non-null")) );	}	TupleDesc tuple_desc;	if( get_call_result_type( fcinfo, NULL, &tuple_desc ) != TYPEFUNC_COMPOSITE ) {		ereport( ERROR, (errmsg("summarize_variant: function returning composite type called in context that cannot accept composite type")) );	}	tuple_desc = BlessTupleDesc( tuple_desc );	ArrayType* values = PG_GETARG_ARRAYTYPE_P( 0 );	Oid values_type = ARR_ELEMTYPE( values );	int16 values_width;	bool values_passbyvalue;	char values_alignmentcode;	Datum* values_content;	bool* values_nullflags;	int values_length;	get_typlenbyvalalign( values_type, &values_width, &values_passbyvalue, &values_alignmentcode );	deconstruct_array( values, values_type, values_width, values_passbyvalue, values_alignmentcode, &values_content, &values_nullflags, &values_length );	size_t composite_type_elements = 9;	Datum* results_content = (Datum*)palloc0( sizeof(Datum) * composite_type_elements );	bool* results_nullflags = (bool*)palloc0( sizeof(bool) * composite_type_elements );	// FORMAT	// [0] - call rate	// [1] - subset call rate	// [2] - alternate allele frequency	// [3] - sample alternate allele frequency	if( !PG_ARGISNULL(1) ) {		ArrayType* indices = PG_GETARG_ARRAYTYPE_P(1);		Oid indices_type = ARR_ELEMTYPE( indices );		int16 indices_width;		bool indices_passbyvalue;		char indices_alignmentcode;		Datum* indices_content;		bool* indices_nullflags;		int indices_length;		get_typlenbyvalalign( indices_type, &indices_width, &indices_passbyvalue, &indices_alignmentcode );		deconstruct_array( indices, indices_type, indices_width, indices_passbyvalue, indices_alignmentcode, &indices_content, &indices_nullflags, &indices_length );		int count = 0;		int nonnull_count = 0;		int alternate_count = 0;		int i;		for( i = 0; i < indices_length; ++i ) {			if( !indices_nullflags[i] && indices_content[i] - 1 < (long unsigned)values_length ) {				++count;				if( !values_nullflags[ indices_content[i] - 1 ] ) {					++nonnull_count;					alternate_count += values_content[ indices_content[i] - 1 ];				}			}		}		results_content[1] = Float4GetDatum( nonnull_count / (float4)count );		results_content[3] = Float4GetDatum( nonnull_count == 0 ? 0 : alternate_count/(2.0*nonnull_count) );		results_nullflags[3] = nonnull_count == 0;	} else {		results_nullflags[1] = true;		results_nullflags[3] = true;	}	int count = values_length;	unsigned int nonnull_count = 0;	unsigned int alternate_count = 0;	int i;	for( i = 0; i < values_length; ++i ) {		if( !values_nullflags[i] ) {			++nonnull_count;			alternate_count += values_content[i];		}	}	results_content[0] = Float4GetDatum( nonnull_count / (float4)count );	results_content[2] = Float4GetDatum( nonnull_count == 0 ? 0 : alternate_count/(2.0*nonnull_count) );	results_nullflags[2] = nonnull_count == 0;	HeapTuple tuple = heap_form_tuple( tuple_desc, results_content, results_nullflags );	Datum result = HeapTupleGetDatum( tuple );	PG_RETURN_DATUM( result );}

/* * SQL function json_array_elements * * get the elements from a json array * * a lot of this processing is similar to the json_each* functions */Datumjson_array_elements(PG_FUNCTION_ARGS){	text	   *json = PG_GETARG_TEXT_P(0);	/* elements doesn't need any escaped strings, so use false here */	JsonLexContext *lex = makeJsonLexContext(json, false);	JsonSemAction sem;	ReturnSetInfo *rsi;	MemoryContext old_cxt;	TupleDesc	tupdesc;	ElementsState state;	state = palloc0(sizeof(elementsState));	sem = palloc0(sizeof(jsonSemAction));	rsi = (ReturnSetInfo *) fcinfo->resultinfo;	if (!rsi || !IsA(rsi, ReturnSetInfo) ||		(rsi->allowedModes & SFRM_Materialize) == 0 ||		rsi->expectedDesc == NULL)		ereport(ERROR,				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),				 errmsg("set-valued function called in context that "						"cannot accept a set")));	rsi->returnMode = SFRM_Materialize;	/* it's a simple type, so don't use get_call_result_type() */	tupdesc = rsi->expectedDesc;	/* make these in a sufficiently long-lived memory context */	old_cxt = MemoryContextSwitchTo(rsi->econtext->ecxt_per_query_memory);	state->ret_tdesc = CreateTupleDescCopy(tupdesc);	BlessTupleDesc(state->ret_tdesc);	state->tuple_store =		tuplestore_begin_heap(rsi->allowedModes & SFRM_Materialize,							  false, work_mem);	MemoryContextSwitchTo(old_cxt);	sem->semstate = (void *) state;	sem->object_start = elements_object_start;	sem->scalar = elements_scalar;	sem->array_element_start = elements_array_element_start;	sem->array_element_end = elements_array_element_end;	state->lex = lex;	state->tmp_cxt = AllocSetContextCreate(CurrentMemoryContext,										 "json_array_elements temporary cxt",										   ALLOCSET_DEFAULT_MINSIZE,										   ALLOCSET_DEFAULT_INITSIZE,										   ALLOCSET_DEFAULT_MAXSIZE);	pg_parse_json(lex, sem);	rsi->setResult = state->tuple_store;	rsi->setDesc = state->ret_tdesc;	PG_RETURN_NULL();}

//.........这里部分代码省略.........  TupleDesc            tuple_desc;  apsp_element_t      *pair;                           /* stuff done only on the first call of the function */  if (SRF_IS_FIRSTCALL())    {      MemoryContext   oldcontext;      int pair_count = 0;        int ret;      /* create a function context for cross-call persistence */      funcctx = SRF_FIRSTCALL_INIT();      /* switch to memory context appropriate for multiple function calls */      oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);      ret = compute_apsp_warshall(text2char(PG_GETARG_TEXT_P(0)),                                  PG_GETARG_BOOL(1),                                  PG_GETARG_BOOL(2), &pair, &pair_count);                  #ifdef DEBUG      DBG("Ret is %i", ret);      if (ret >= 0)         {          int i;          for (i = 0; i < pair_count; i++)             {              DBG("Step: %i, source_id: %i, target_id: %i, cost: %f ", i, pair[i].src_vertex_id, pair[i].dest_vertex_id, pair[i].cost);            }        }#endif      /* total number of tuples to be returned */      funcctx->max_calls = pair_count;      funcctx->user_fctx = pair;      funcctx->tuple_desc =         BlessTupleDesc(RelationNameGetTupleDesc("pgr_costResult"));      MemoryContextSwitchTo(oldcontext);    }  /* stuff done on every call of the function */  funcctx = SRF_PERCALL_SETUP();  call_cntr = funcctx->call_cntr;  max_calls = funcctx->max_calls;  tuple_desc = funcctx->tuple_desc;  pair = (apsp_element_t*) funcctx->user_fctx;  if (call_cntr < max_calls)    /* do when there is more left to send */    {      HeapTuple    tuple;      Datum        result;      Datum *values;      char* nulls;      /* This will work for some compilers. If it crashes with segfault, try to change the following block with this one           values = palloc(4 * sizeof(Datum));      nulls = palloc(4 * sizeof(char));        values[0] = call_cntr;      nulls[0] = ' ';      values[1] = Int32GetDatum(path[call_cntr].vertex_id);      nulls[1] = ' ';      values[2] = Int32GetDatum(path[call_cntr].edge_id);      nulls[2] = ' ';      values[3] = Float8GetDatum(path[call_cntr].cost);      nulls[3] = ' ';      */          values = palloc(4 * sizeof(Datum));      nulls = palloc(4 * sizeof(char));      values[0] = Int32GetDatum(call_cntr);      nulls[0] = ' ';      values[1] = Int32GetDatum(pair[call_cntr].src_vertex_id);      nulls[1] = ' ';      values[2] = Int32GetDatum(pair[call_cntr].dest_vertex_id);      nulls[2] = ' ';      values[3] = Float8GetDatum(pair[call_cntr].cost);      nulls[3] = ' ';		            tuple = heap_formtuple(tuple_desc, values, nulls);      /* make the tuple into a datum */      result = HeapTupleGetDatum(tuple);      /* clean up (this is not really necessary) */      pfree(values);      pfree(nulls);      SRF_RETURN_NEXT(funcctx, result);    }  else    /* do when there is no more left */    {      SRF_RETURN_DONE(funcctx);    }}

static inline Datumeach_worker(PG_FUNCTION_ARGS, bool as_text){	text	   *json = PG_GETARG_TEXT_P(0);	JsonLexContext *lex = makeJsonLexContext(json, true);	JsonSemAction sem;	ReturnSetInfo *rsi;	MemoryContext old_cxt;	TupleDesc	tupdesc;	EachState	state;	state = palloc0(sizeof(eachState));	sem = palloc0(sizeof(jsonSemAction));	rsi = (ReturnSetInfo *) fcinfo->resultinfo;	if (!rsi || !IsA(rsi, ReturnSetInfo) ||		(rsi->allowedModes & SFRM_Materialize) == 0 ||		rsi->expectedDesc == NULL)		ereport(ERROR,				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),				 errmsg("set-valued function called in context that "						"cannot accept a set")));	rsi->returnMode = SFRM_Materialize;	(void) get_call_result_type(fcinfo, NULL, &tupdesc);	/* make these in a sufficiently long-lived memory context */	old_cxt = MemoryContextSwitchTo(rsi->econtext->ecxt_per_query_memory);	state->ret_tdesc = CreateTupleDescCopy(tupdesc);	BlessTupleDesc(state->ret_tdesc);	state->tuple_store =		tuplestore_begin_heap(rsi->allowedModes & SFRM_Materialize,							  false, work_mem);	MemoryContextSwitchTo(old_cxt);	sem->semstate = (void *) state;	sem->array_start = each_array_start;	sem->scalar = each_scalar;	sem->object_field_start = each_object_field_start;	sem->object_field_end = each_object_field_end;	state->normalize_results = as_text;	state->next_scalar = false;	state->lex = lex;	state->tmp_cxt = AllocSetContextCreate(CurrentMemoryContext,										   "json_each temporary cxt",										   ALLOCSET_DEFAULT_MINSIZE,										   ALLOCSET_DEFAULT_INITSIZE,										   ALLOCSET_DEFAULT_MAXSIZE);	pg_parse_json(lex, sem);	rsi->setResult = state->tuple_store;	rsi->setDesc = state->ret_tdesc;	PG_RETURN_NULL();}

/* * gp_read_error_log * * Returns set of error log tuples. */Datumgp_read_error_log(PG_FUNCTION_ARGS){	FuncCallContext	   *funcctx;	ReadErrorLogContext *context;	HeapTuple			tuple;	Datum				result;	/*	 * First call setup	 */	if (SRF_IS_FIRSTCALL())	{		MemoryContext	oldcontext;		FILE	   *fp;		text	   *relname;		funcctx = SRF_FIRSTCALL_INIT();		relname = PG_GETARG_TEXT_P(0);		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);		context = palloc0(sizeof(ReadErrorLogContext));		funcctx->user_fctx = (void *) context;		funcctx->tuple_desc = BlessTupleDesc(GetErrorTupleDesc());		/*		 * Though this function is usually executed on segment, we dispatch		 * the execution if it happens to be on QD, and combine the results		 * into one set.		 */		if (Gp_role == GP_ROLE_DISPATCH)		{			struct CdbPgResults cdb_pgresults = {NULL, 0};			StringInfoData sql;			int		i;			initStringInfo(&sql);			/*			 * construct SQL			 */			appendStringInfo(&sql,					"SELECT * FROM pg_catalog.gp_read_error_log(%s) ",							 quote_literal_internal(text_to_cstring(relname)));			CdbDispatchCommand(sql.data, DF_WITH_SNAPSHOT, &cdb_pgresults);			for (i = 0; i < cdb_pgresults.numResults; i++)			{				if (PQresultStatus(cdb_pgresults.pg_results[i]) != PGRES_TUPLES_OK)				{					cdbdisp_clearCdbPgResults(&cdb_pgresults);					elog(ERROR, "unexpected result from segment: %d",								PQresultStatus(cdb_pgresults.pg_results[i]));				}				context->numTuples += PQntuples(cdb_pgresults.pg_results[i]);			}			pfree(sql.data);			context->segResults = cdb_pgresults.pg_results;			context->numSegResults = cdb_pgresults.numResults;		}		else		{			/*			 * In QE, read the error log.			 */			RangeVar	   *relrv;			Oid				relid;			relrv = makeRangeVarFromNameList(textToQualifiedNameList(relname));			relid = RangeVarGetRelid(relrv, true);			/*			 * If the relation has gone, silently return no tuples.			 */			if (OidIsValid(relid))			{				AclResult aclresult;				/*				 * Requires SELECT priv to read error log.				 */				aclresult = pg_class_aclcheck(relid, GetUserId(), ACL_SELECT);				if (aclresult != ACLCHECK_OK)					aclcheck_error(aclresult, ACL_KIND_CLASS, relrv->relname);				ErrorLogFileName(context->filename, MyDatabaseId, relid);				fp = AllocateFile(context->filename, "r");				context->fp = fp;			}		}//.........这里部分代码省略.........

/* * pg_lock_status - produce a view with one row per held or awaited lock mode */Datumpg_lock_status(PG_FUNCTION_ARGS){	FuncCallContext *funcctx;	PG_Lock_Status *mystatus;	LockData   *lockData;	if (SRF_IS_FIRSTCALL())	{		TupleDesc	tupdesc;		MemoryContext oldcontext;		/* create a function context for cross-call persistence */		funcctx = SRF_FIRSTCALL_INIT();		/*		 * switch to memory context appropriate for multiple function calls		 */		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);		/* build tupdesc for result tuples */		/* this had better match pg_locks view in system_views.sql */		tupdesc = CreateTemplateTupleDesc(16, false);		TupleDescInitEntry(tupdesc, (AttrNumber) 1, "locktype",						   TEXTOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 2, "database",						   OIDOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 3, "relation",						   OIDOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 4, "page",						   INT4OID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 5, "tuple",						   INT2OID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 6, "transactionid",						   XIDOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 7, "classid",						   OIDOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 8, "objid",						   OIDOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 9, "objsubid",						   INT2OID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 10, "transaction",						   XIDOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 11, "pid",						   INT4OID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 12, "mode",						   TEXTOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 13, "granted",						   BOOLOID, -1, 0);		/*		 * These next columns are specific to GPDB		 */		TupleDescInitEntry(tupdesc, (AttrNumber) 14, "mppSessionId",						   INT4OID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 15, "mppIsWriter",						   BOOLOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 16, "gp_segment_id",						   INT4OID, -1, 0);		funcctx->tuple_desc = BlessTupleDesc(tupdesc);		/*		 * Collect all the locking information that we will format and send		 * out as a result set.		 */		mystatus = (PG_Lock_Status *) palloc(sizeof(PG_Lock_Status));		funcctx->user_fctx = (void *) mystatus;		mystatus->lockData = GetLockStatusData();		mystatus->currIdx = 0;		mystatus->numSegLocks = 0;		mystatus->numsegresults = 0;		mystatus->segresults = NULL;	}	funcctx = SRF_PERCALL_SETUP();	mystatus = (PG_Lock_Status *) funcctx->user_fctx;	lockData = mystatus->lockData;	/*	 * This loop returns all the local lock data from the segment we are running on.	 */	while (mystatus->currIdx < lockData->nelements)	{		PROCLOCK   *proclock;		LOCK	   *lock;		PGPROC	   *proc;		bool		granted;		LOCKMODE	mode = 0;		const char *locktypename;		char		tnbuf[32];		Datum		values[16];		bool		nulls[16];		HeapTuple	tuple;		Datum		result;//.........这里部分代码省略.........

/* * master_get_local_first_candidate_nodes returns a set of candidate host names * and port numbers on which to place new shards. The function makes sure to * always allocate the first candidate node as the node the caller is connecting * from; and allocates additional nodes until the shard replication factor is * met. The function errors if the caller's remote node name is not found in the * membership list, or if the number of available nodes falls short of the * replication factor. */Datummaster_get_local_first_candidate_nodes(PG_FUNCTION_ARGS){	FuncCallContext *functionContext = NULL;	uint32 desiredNodeCount = 0;	uint32 currentNodeCount = 0;	if (SRF_IS_FIRSTCALL())	{		MemoryContext oldContext = NULL;		TupleDesc tupleDescriptor = NULL;		uint32 liveNodeCount = 0;		bool hasOid = false;		/* create a function context for cross-call persistence */		functionContext = SRF_FIRSTCALL_INIT();		/* switch to memory context appropriate for multiple function calls */		oldContext = MemoryContextSwitchTo(functionContext->multi_call_memory_ctx);		functionContext->user_fctx = NIL;		functionContext->max_calls = ShardReplicationFactor;		/* if enough live nodes, return an extra candidate node as backup */		liveNodeCount = WorkerGetLiveNodeCount();		if (liveNodeCount > ShardReplicationFactor)		{			functionContext->max_calls = ShardReplicationFactor + 1;		}		/*		 * This tuple descriptor must match the output parameters declared for		 * the function in pg_proc.		 */		tupleDescriptor = CreateTemplateTupleDesc(CANDIDATE_NODE_FIELDS, hasOid);		TupleDescInitEntry(tupleDescriptor, (AttrNumber) 1, "node_name",						   TEXTOID, -1, 0);		TupleDescInitEntry(tupleDescriptor, (AttrNumber) 2, "node_port",						   INT8OID, -1, 0);		functionContext->tuple_desc = BlessTupleDesc(tupleDescriptor);		MemoryContextSwitchTo(oldContext);	}	functionContext = SRF_PERCALL_SETUP();	desiredNodeCount = functionContext->max_calls;	currentNodeCount = functionContext->call_cntr;	if (currentNodeCount < desiredNodeCount)	{		MemoryContext oldContext = NULL;		List *currentNodeList = NIL;		WorkerNode *candidateNode = NULL;		Datum candidateDatum = 0;		/* switch to memory context appropriate for multiple function calls */		oldContext = MemoryContextSwitchTo(functionContext->multi_call_memory_ctx);		currentNodeList = functionContext->user_fctx;		candidateNode = WorkerGetLocalFirstCandidateNode(currentNodeList);		if (candidateNode == NULL)		{			ereport(ERROR, (errmsg("could only find %u of %u required nodes",								   currentNodeCount, desiredNodeCount)));		}		currentNodeList = lappend(currentNodeList, candidateNode);		functionContext->user_fctx = currentNodeList;		MemoryContextSwitchTo(oldContext);		candidateDatum = WorkerNodeGetDatum(candidateNode, functionContext->tuple_desc);		SRF_RETURN_NEXT(functionContext, candidateDatum);	}	else	{		SRF_RETURN_DONE(functionContext);	}}

/* * pg_prepared_xact *		Produce a view with one row per prepared transaction. * * This function is here so we don't have to export the * GlobalTransactionData struct definition. */Datumpg_prepared_xact(PG_FUNCTION_ARGS){	FuncCallContext *funcctx;	Working_State *status;	if (SRF_IS_FIRSTCALL())	{		TupleDesc	tupdesc;		MemoryContext oldcontext;		/* create a function context for cross-call persistence */		funcctx = SRF_FIRSTCALL_INIT();		/*		 * Switch to memory context appropriate for multiple function calls		 */		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);		/* build tupdesc for result tuples */		/* this had better match pg_prepared_xacts view in system_views.sql */		tupdesc = CreateTemplateTupleDesc(5, false);		TupleDescInitEntry(tupdesc, (AttrNumber) 1, "transaction",						   XIDOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 2, "gid",						   TEXTOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 3, "prepared",						   TIMESTAMPTZOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 4, "ownerid",						   OIDOID, -1, 0);		TupleDescInitEntry(tupdesc, (AttrNumber) 5, "dbid",						   OIDOID, -1, 0);		funcctx->tuple_desc = BlessTupleDesc(tupdesc);		/*		 * Collect all the 2PC status information that we will format and send		 * out as a result set.		 */		status = (Working_State *) palloc(sizeof(Working_State));		funcctx->user_fctx = (void *) status;		status->ngxacts = GetPreparedTransactionList(&status->array);		status->currIdx = 0;		MemoryContextSwitchTo(oldcontext);	}	funcctx = SRF_PERCALL_SETUP();	status = (Working_State *) funcctx->user_fctx;	while (status->array != NULL && status->currIdx < status->ngxacts)	{		GlobalTransaction gxact = &status->array[status->currIdx++];		Datum		values[5];		bool		nulls[5];		HeapTuple	tuple;		Datum		result;		if (!gxact->valid)			continue;		/*		 * Form tuple with appropriate data.		 */		MemSet(values, 0, sizeof(values));		MemSet(nulls, 0, sizeof(nulls));		values[0] = TransactionIdGetDatum(gxact->proc.xid);		values[1] = CStringGetTextDatum(gxact->gid);		values[2] = TimestampTzGetDatum(gxact->prepared_at);		values[3] = ObjectIdGetDatum(gxact->owner);		values[4] = ObjectIdGetDatum(gxact->proc.databaseId);		tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);		result = HeapTupleGetDatum(tuple);		SRF_RETURN_NEXT(funcctx, result);	}	SRF_RETURN_DONE(funcctx);}

/* * topn is a user-facing UDF which returns the top items and their frequencies. * It first gets the top-n structure and converts it into the ordered array of * FrequentTopnItem which keeps Datums and the frequencies in the first call. * Then, it returns an item and its frequency according to call counter. This * function requires a parameter for the type because PostgreSQL has strongly * typed system and the type of frequent items in returning rows has to be given. */Datumtopn(PG_FUNCTION_ARGS){	FuncCallContext *functionCallContext = NULL;	TupleDesc tupleDescriptor = NULL;	Oid returningItemType = get_fn_expr_argtype(fcinfo->flinfo, 1);	CmsTopn *cmsTopn = NULL;	ArrayType *topnArray = NULL;	int topnArrayLength = 0;	Datum topnItem = 0;	bool isNull = false;	ArrayIterator topnIterator = NULL;	bool hasMoreItem = false;	int callCounter = 0;	int maxCallCounter = 0;	if (SRF_IS_FIRSTCALL())	{		MemoryContext oldcontext = NULL;		Size topnArraySize = 0;		TypeCacheEntry *itemTypeCacheEntry = NULL;		int topnIndex = 0;		Oid itemType = InvalidOid;		FrequentTopnItem *sortedTopnArray = NULL;		TupleDesc completeTupleDescriptor = NULL;		functionCallContext = SRF_FIRSTCALL_INIT();		if (PG_ARGISNULL(0))		{			SRF_RETURN_DONE(functionCallContext);		}		cmsTopn = (CmsTopn *)  PG_GETARG_VARLENA_P(0);		topnArray = TopnArray(cmsTopn);		topnArrayLength = ARR_DIMS(topnArray)[0];		/* if there is not any element in the array just return */		if (topnArrayLength == 0)		{			SRF_RETURN_DONE(functionCallContext);		}		itemType = ARR_ELEMTYPE(topnArray);		if (itemType != returningItemType)		{			elog(ERROR, "not a proper cms_topn for the result type");		}		/* switch to consistent context for multiple calls */		oldcontext = MemoryContextSwitchTo(functionCallContext->multi_call_memory_ctx);		itemTypeCacheEntry = lookup_type_cache(itemType, 0);		functionCallContext->max_calls = topnArrayLength;		/* create an array to copy top-n items and sort them later */		topnArraySize = sizeof(FrequentTopnItem) * topnArrayLength;		sortedTopnArray = palloc0(topnArraySize);		topnIterator = array_create_iterator(topnArray, 0);		hasMoreItem = array_iterate(topnIterator, &topnItem, &isNull);		while (hasMoreItem)		{			FrequentTopnItem frequentTopnItem;			frequentTopnItem.topnItem = topnItem;			frequentTopnItem.topnItemFrequency =			        CmsTopnEstimateItemFrequency(cmsTopn, topnItem, itemTypeCacheEntry);			sortedTopnArray[topnIndex] = frequentTopnItem;			hasMoreItem = array_iterate(topnIterator, &topnItem, &isNull);			topnIndex++;		}		SortTopnItems(sortedTopnArray, topnArrayLength);		functionCallContext->user_fctx = sortedTopnArray;		get_call_result_type(fcinfo, &returningItemType, &tupleDescriptor);		completeTupleDescriptor = BlessTupleDesc(tupleDescriptor);		functionCallContext->tuple_desc = completeTupleDescriptor;		MemoryContextSwitchTo(oldcontext);	}	functionCallContext = SRF_PERCALL_SETUP();	maxCallCounter = functionCallContext->max_calls;	callCounter = functionCallContext->call_cntr;	if (callCounter < maxCallCounter)	{		Datum *tupleValues = (Datum *) palloc(2 * sizeof(Datum));		HeapTuple topnItemTuple;		Datum topnItemDatum = 0;		char *tupleNulls = (char *) palloc0(2 * sizeof(char));		FrequentTopnItem *sortedTopnArray = NULL;		TupleDesc completeTupleDescriptor = NULL;//.........这里部分代码省略.........