/*
 * deflist_to_tuplestore - Helper function to convert DefElem list to
 * tuplestore usable in SRF.
 */
static void
deflist_to_tuplestore(ReturnSetInfo *rsinfo, List *options)
{
	ListCell   *cell;
	TupleDesc	tupdesc;
	Tuplestorestate *tupstore;
	Datum		values[2];
	bool		nulls[2];
	MemoryContext per_query_ctx;
	MemoryContext oldcontext;

	/* check to see if caller supports us returning a tuplestore */
	if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("set-valued function called in context that cannot accept a set")));
	if (!(rsinfo->allowedModes & SFRM_Materialize) ||
		rsinfo->expectedDesc == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("materialize mode required, but it is not allowed in this context")));

	per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
	oldcontext = MemoryContextSwitchTo(per_query_ctx);

	/*
	 * Now prepare the result set.
	 */
	tupdesc = CreateTupleDescCopy(rsinfo->expectedDesc);
	tupstore = tuplestore_begin_heap(true, false, work_mem);
	rsinfo->returnMode = SFRM_Materialize;
	rsinfo->setResult = tupstore;
	rsinfo->setDesc = tupdesc;

	foreach(cell, options)
	{
		DefElem    *def = lfirst(cell);

		values[0] = CStringGetTextDatum(def->defname);
		nulls[0] = false;
		if (def->arg)
		{
			values[1] = CStringGetTextDatum(((Value *) (def->arg))->val.str);
			nulls[1] = false;
		}
		else
		{
			values[1] = (Datum) 0;
			nulls[1] = true;
		}
		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
	}

/*
 * pgstrom_debug_info
 *
 * shows user's debug information
 */
Datum
pgstrom_debug_info(PG_FUNCTION_ARGS)
{
	FuncCallContext	   *fncxt;
	MemoryContext		oldcxt;
	ListCell   *cell;
	DefElem	   *defel;
	Datum		values[2];
	bool		isnull[2];
	HeapTuple	tuple;

	if (SRF_IS_FIRSTCALL())
	{
		TupleDesc	tupdesc;
		List	   *debug_info_list = NIL;

		fncxt = SRF_FIRSTCALL_INIT();

		oldcxt = MemoryContextSwitchTo(fncxt->multi_call_memory_ctx);

		tupdesc = CreateTemplateTupleDesc(2, false);
		TupleDescInitEntry(tupdesc, (AttrNumber) 1, "key",
						   TEXTOID, -1, 0);
		TupleDescInitEntry(tupdesc, (AttrNumber) 2, "value",
						   TEXTOID, -1, 0);

		debug_info_list = pgstrom_scan_debug_info(debug_info_list);

		fncxt->user_fctx = (void *) debug_info_list;
		fncxt->tuple_desc = BlessTupleDesc(tupdesc);

		MemoryContextSwitchTo(oldcxt);
	}
	fncxt = SRF_PERCALL_SETUP();
	cell = list_head((List *)fncxt->user_fctx);
	if (!cell)
		SRF_RETURN_DONE(fncxt);

	defel = lfirst(cell);
	Assert(IsA(defel, DefElem));

	memset(isnull, false, sizeof(isnull));
	values[0] = CStringGetTextDatum(defel->defname);
	values[1] = CStringGetTextDatum(strVal(defel->arg));

	tuple = heap_form_tuple(fncxt->tuple_desc, values, isnull);

	fncxt->user_fctx = list_delete_ptr((List *)fncxt->user_fctx,
									   lfirst(cell));
	SRF_RETURN_NEXT(fncxt, HeapTupleGetDatum(tuple));
}

/*
 * LogTransactionRecord registers the fact that a transaction has been
 * prepared on a worker. The presence of this record indicates that the
 * prepared transaction should be committed.
 */
void
LogTransactionRecord(int groupId, char *transactionName)
{
	Relation pgDistTransaction = NULL;
	TupleDesc tupleDescriptor = NULL;
	HeapTuple heapTuple = NULL;
	Datum values[Natts_pg_dist_transaction];
	bool isNulls[Natts_pg_dist_transaction];

	/* form new transaction tuple */
	memset(values, 0, sizeof(values));
	memset(isNulls, false, sizeof(isNulls));

	values[Anum_pg_dist_transaction_groupid - 1] = Int32GetDatum(groupId);
	values[Anum_pg_dist_transaction_gid - 1] = CStringGetTextDatum(transactionName);

	/* open transaction relation and insert new tuple */
	pgDistTransaction = heap_open(DistTransactionRelationId(), RowExclusiveLock);

	tupleDescriptor = RelationGetDescr(pgDistTransaction);
	heapTuple = heap_form_tuple(tupleDescriptor, values, isNulls);

	simple_heap_insert(pgDistTransaction, heapTuple);
	CatalogUpdateIndexes(pgDistTransaction, heapTuple);
	CommandCounterIncrement();

	/* close relation and invalidate previous cache entry */
	heap_close(pgDistTransaction, RowExclusiveLock);
}

/*
 * master_metadata_snapshot prints all the queries that are required
 * to generate a metadata snapshot.
 */
Datum
master_metadata_snapshot(PG_FUNCTION_ARGS)
{
	List *dropSnapshotCommands = MetadataDropCommands();
	List *createSnapshotCommands = MetadataCreateCommands();
	List *snapshotCommandList = NIL;
	ListCell *snapshotCommandCell = NULL;
	int snapshotCommandCount = 0;
	Datum *snapshotCommandDatumArray = NULL;
	ArrayType *snapshotCommandArrayType = NULL;
	int snapshotCommandIndex = 0;
	Oid ddlCommandTypeId = TEXTOID;

	snapshotCommandList = list_concat(snapshotCommandList, dropSnapshotCommands);
	snapshotCommandList = list_concat(snapshotCommandList, createSnapshotCommands);

	snapshotCommandCount = list_length(snapshotCommandList);
	snapshotCommandDatumArray = palloc0(snapshotCommandCount * sizeof(Datum));

	foreach(snapshotCommandCell, snapshotCommandList)
	{
		char *metadataSnapshotCommand = (char *) lfirst(snapshotCommandCell);
		Datum metadataSnapshotCommandDatum = CStringGetTextDatum(metadataSnapshotCommand);

		snapshotCommandDatumArray[snapshotCommandIndex] = metadataSnapshotCommandDatum;
		snapshotCommandIndex++;
	}

Datum
plvsubst_string_string(PG_FUNCTION_ARGS)
{
	Datum		r;
	ArrayType  *array;
	FunctionCallInfoData locfcinfo;

	init_c_subst();

	if (PG_ARGISNULL(0) || PG_ARGISNULL(1))
		PG_RETURN_NULL();

	/*
	 * I can't use DirectFunctionCall2
	 */

	InitFunctionCallInfoData(locfcinfo, fcinfo->flinfo, 2, NULL, NULL);
	locfcinfo.arg[0] = PG_GETARG_DATUM(1);
	locfcinfo.arg[1] = PG_GETARG_IF_EXISTS(2, DATUM, CStringGetTextDatum(","));
	locfcinfo.argnull[0] = false;
	locfcinfo.argnull[1] = false;
	r = text_to_array(&locfcinfo);

	if (locfcinfo.isnull || r == (Datum) 0)
		array = NULL;
	else
		array = DatumGetArrayTypeP(r);

	PG_RETURN_TEXT_P(plvsubst_string(PG_GETARG_TEXT_P(0),
					 array,
					 PG_GETARG_IF_EXISTS(3, TEXT_P, c_subst),
					 fcinfo));
}

static uint16
GetNewLabelId(char *graphname, Oid graphid)
{
	char		sname[128];
	Datum		stext;
	uint16		labid;
	int			cnt;

	snprintf(sname, 128, "%s.%s", graphname, AG_LABEL_SEQ);
	stext = CStringGetTextDatum(sname);

	cnt = 0;
	for (;;)
	{
		Datum val;

		val = DirectFunctionCall1(nextval, stext);
		labid = DatumGetUInt16(val);
		if (!labid_exists(graphid, labid))
			break;

		if (++cnt >= GRAPHID_LABID_MAX)
			ereport(ERROR,
					(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
					 errmsg("no more new labels are available")));
	}

	return labid;
}

Datum
brin_metapage_info(PG_FUNCTION_ARGS)
{
    bytea	   *raw_page = PG_GETARG_BYTEA_P(0);
    Page		page;
    BrinMetaPageData *meta;
    TupleDesc	tupdesc;
    Datum		values[4];
    bool		nulls[4];
    HeapTuple	htup;

    page = verify_brin_page(raw_page, BRIN_PAGETYPE_META, "metapage");

    /* Build a tuple descriptor for our result type */
    if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
        elog(ERROR, "return type must be a row type");
    tupdesc = BlessTupleDesc(tupdesc);

    /* Extract values from the metapage */
    meta = (BrinMetaPageData *) PageGetContents(page);
    MemSet(nulls, 0, sizeof(nulls));
    values[0] = CStringGetTextDatum(psprintf("0x%08X", meta->brinMagic));
    values[1] = Int32GetDatum(meta->brinVersion);
    values[2] = Int32GetDatum(meta->pagesPerRange);
    values[3] = Int64GetDatum(meta->lastRevmapPage);

    htup = heap_form_tuple(tupdesc, values, nulls);

    PG_RETURN_DATUM(HeapTupleGetDatum(htup));
}

static void
each_object_field_end(void *state, char *fname, bool isnull)
{
	EachState	_state = (EachState) state;
	MemoryContext old_cxt;
	int			len;
	text	   *val;
	HeapTuple	tuple;
	Datum		values[2];
	bool		nulls[2] = {false, false};

	/* skip over nested objects */
	if (_state->lex->lex_level != 1)
		return;

	/* use the tmp context so we can clean up after each tuple is done */
	old_cxt = MemoryContextSwitchTo(_state->tmp_cxt);

	values[0] = CStringGetTextDatum(fname);

	if (isnull && _state->normalize_results)
	{
		nulls[1] = true;
		values[1] = (Datum) NULL;
	}
	else if (_state->next_scalar)
	{
		values[1] = CStringGetTextDatum(_state->normalized_scalar);
		_state->next_scalar = false;
	}
	else
	{
		len = _state->lex->prev_token_terminator - _state->result_start;
		val = cstring_to_text_with_len(_state->result_start, len);
		values[1] = PointerGetDatum(val);
	}


	tuple = heap_form_tuple(_state->ret_tdesc, values, nulls);

	tuplestore_puttuple(_state->tuple_store, tuple);

	/* clean up and switch back */
	MemoryContextSwitchTo(old_cxt);
	MemoryContextReset(_state->tmp_cxt);
}

/*
 * SQL function for creating a new logical replication slot.
 */
Datum
pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
{
	Name		name = PG_GETARG_NAME(0);
	Name		plugin = PG_GETARG_NAME(1);

	LogicalDecodingContext *ctx = NULL;

	TupleDesc	tupdesc;
	HeapTuple	tuple;
	Datum		result;
	Datum		values[2];
	bool		nulls[2];

	Assert(!MyReplicationSlot);

	if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
		elog(ERROR, "return type must be a row type");

	check_permissions();

	CheckLogicalDecodingRequirements();

	/*
	 * Acquire a logical decoding slot, this will check for conflicting names.
	 * Initially create it as ephemeral - that allows us to nicely handle
	 * errors during initialization because it'll get dropped if this
	 * transaction fails. We'll make it persistent at the end.
	 */
	ReplicationSlotCreate(NameStr(*name), true, RS_EPHEMERAL);

	/*
	 * Create logical decoding context, to build the initial snapshot.
	 */
	ctx = CreateInitDecodingContext(
									NameStr(*plugin), NIL,
									logical_read_local_xlog_page, NULL, NULL);

	/* build initial snapshot, might take a while */
	DecodingContextFindStartpoint(ctx);

	values[0] = CStringGetTextDatum(NameStr(MyReplicationSlot->data.name));
	values[1] = LSNGetDatum(MyReplicationSlot->data.confirmed_flush);

	/* don't need the decoding context anymore */
	FreeDecodingContext(ctx);

	memset(nulls, 0, sizeof(nulls));

	tuple = heap_form_tuple(tupdesc, values, nulls);
	result = HeapTupleGetDatum(tuple);

	/* ok, slot is now fully created, mark it as persistent */
	ReplicationSlotPersist();
	ReplicationSlotRelease();

	PG_RETURN_DATUM(result);
}

/*
* Enables transformations in the optimizer.
*/
Datum
enable_xform(PG_FUNCTION_ARGS)
{
#ifdef USE_ORCA
	return EnableXform(fcinfo);
#else
	return CStringGetTextDatum("ORCA not supported");
#endif
}

/*
 * Compute an xlog file name and decimal byte offset given a WAL location,
 * such as is returned by pg_stop_backup() or pg_xlog_switch().
 *
 * Note that a location exactly at a segment boundary is taken to be in
 * the previous segment.  This is usually the right thing, since the
 * expected usage is to determine which xlog file(s) are ready to archive.
 */
Datum
pg_xlogfile_name_offset(PG_FUNCTION_ARGS)
{
	XLogSegNo	xlogsegno;
	uint32		xrecoff;
	XLogRecPtr	locationpoint = PG_GETARG_LSN(0);
	char		xlogfilename[MAXFNAMELEN];
	Datum		values[2];
	bool		isnull[2];
	TupleDesc	resultTupleDesc;
	HeapTuple	resultHeapTuple;
	Datum		result;

	if (RecoveryInProgress())
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("recovery is in progress"),
				 errhint("pg_xlogfile_name_offset() cannot be executed during recovery.")));

	/*
	 * Construct a tuple descriptor for the result row.  This must match this
	 * function's pg_proc entry!
	 */
	resultTupleDesc = CreateTemplateTupleDesc(2, false);
	TupleDescInitEntry(resultTupleDesc, (AttrNumber) 1, "file_name",
					   TEXTOID, -1, 0);
	TupleDescInitEntry(resultTupleDesc, (AttrNumber) 2, "file_offset",
					   INT4OID, -1, 0);

	resultTupleDesc = BlessTupleDesc(resultTupleDesc);

	/*
	 * xlogfilename
	 */
	XLByteToPrevSeg(locationpoint, xlogsegno);
	XLogFileName(xlogfilename, ThisTimeLineID, xlogsegno);

	values[0] = CStringGetTextDatum(xlogfilename);
	isnull[0] = false;

	/*
	 * offset
	 */
	xrecoff = locationpoint % XLogSegSize;

	values[1] = UInt32GetDatum(xrecoff);
	isnull[1] = false;

	/*
	 * Tuple jam: Having first prepared your Datums, then squash together
	 */
	resultHeapTuple = heap_form_tuple(resultTupleDesc, values, isnull);

	result = HeapTupleGetDatum(resultHeapTuple);

	PG_RETURN_DATUM(result);
}

/* 
* Disables transformations in the optimizer.
*/
Datum
disable_xform(PG_FUNCTION_ARGS)
{
#ifdef USE_ORCA
	return DisableXform(fcinfo);
#else
	return CStringGetTextDatum("Pivotal Query Optimizer not supported");
#endif
}

/*
* Returns the optimizer and gpos library versions.
*/
Datum
gp_opt_version(PG_FUNCTION_ARGS __attribute__((unused)))
{
#ifdef USE_ORCA
	return LibraryVersion();
#else
	return CStringGetTextDatum("Pivotal Query Optimizer not supported");
#endif
}

/*
 * pgmpc_lsplaylists
 * List all playlists of remote server.
 */
Datum
pgmpc_lsplaylists(PG_FUNCTION_ARGS)
{
	TupleDesc   tupdesc;
	Tuplestorestate *tupstore;

	/* Initialize function context */
	pgmpc_init_setof_single(fcinfo, TEXTOID, "playlist", &tupdesc, &tupstore);

	/*
	 * Run the command to get all the songs.
	 */
	pgmpc_init();
	if (!mpd_send_list_playlists(mpd_conn))
		pgmpc_print_error();

	/* Now get all the songs and send them back to caller */
	while (true)
	{
		Datum       values[1];
		bool		nulls[1];
		struct mpd_playlist *playlist = mpd_recv_playlist(mpd_conn);

		/* Leave if done */
		if (playlist == NULL)
			break;

		/* Assign song name */
		nulls[0] = false;
		values[0] = CStringGetTextDatum(mpd_playlist_get_path(playlist));

		/* Save values */
		tuplestore_putvalues(tupstore, tupdesc, values, nulls);

		/* Clean up for the next one */
		mpd_playlist_free(playlist);
	}

	/* We may be in error state, so check for it */
	if (mpd_connection_get_error(mpd_conn) != MPD_ERROR_SUCCESS)
	{
		const char *message = mpd_connection_get_error_message(mpd_conn);
		pgmpc_reset();
		ereport(ERROR,
				(errcode(ERRCODE_SYSTEM_ERROR),
				 errmsg("mpd command failed: %s",
						message)));
	}

	/* Clean up */
	pgmpc_reset();

	/* clean up and return the tuplestore */
	tuplestore_donestoring(tupstore);

	return (Datum) 0;
}

/*
 * utl_file_dir security .. is solved with aux. table.
 *
 * Raise exception if don't find string in table.
 */
static void
check_secure_locality(const char *path)
{
	static SPIPlanPtr	plan = NULL;

	Oid		argtypes[] = {TEXTOID};
	Datum	values[1];
	char	nulls[1] = {' '};

	/* hack for availbility regress test */
	if (strcmp(path, "/tmp/regress_orafce") == 0)
		return;

	values[0] = CStringGetTextDatum(path);

	/*
	 * SELECT 1 FROM utl_file.utl_file_dir
	 *   WHERE substring($1, 1, length(dir) + 1) = dir || '/'
	 */

	if (SPI_connect() < 0)
		ereport(ERROR,
			(errcode(ERRCODE_INTERNAL_ERROR),
			 errmsg("SPI_connect failed")));

	if (!plan)
	{
		/* Don't use LIKE not to escape '_' and '%' */
		SPIPlanPtr p = SPI_prepare(
		    "SELECT 1 FROM utl_file.utl_file_dir"
			" WHERE substring($1, 1, length(dir) + 1) = dir || '/'",
		    1, argtypes);

		if (p == NULL || (plan = SPI_saveplan(p)) == NULL)
			ereport(ERROR,
				(errcode(ERRCODE_INTERNAL_ERROR),
				errmsg("SPI_prepare_failed")));
	}

	if (SPI_OK_SELECT != SPI_execute_plan(plan, values, nulls, false, 1))
		ereport(ERROR,
			(errcode(ERRCODE_INTERNAL_ERROR),
			 errmsg("can't execute sql")));

	if (SPI_processed == 0)
		ereport(ERROR,
			(errcode(ERRCODE_RAISE_EXCEPTION),
			 errmsg(INVALID_PATH),
			 errdetail("you cannot access locality"),
			 errhint("locality is not found in utl_file_dir table")));
	SPI_finish();
}

/*
 * VXIDGetDatum - Construct a text representation of a VXID
 *
 * This is currently only used in pg_lock_status, so we put it here.
 */
static Datum
VXIDGetDatum(BackendId bid, LocalTransactionId lxid)
{
	/*
	 * The representation is "<bid>/<lxid>", decimal and unsigned decimal
	 * respectively.  Note that elog.c also knows how to format a vxid.
	 */
	char		vxidstr[32];

	snprintf(vxidstr, sizeof(vxidstr), "%d/%u", bid, lxid);

	return CStringGetTextDatum(vxidstr);
}

/*
 * cms_topn_info returns summary about the given CmsTopn structure.
 */
Datum
cms_topn_info(PG_FUNCTION_ARGS)
{
	CmsTopn *cmsTopn = NULL;
	StringInfo cmsTopnInfoString = makeStringInfo();

	cmsTopn = (CmsTopn *) PG_GETARG_VARLENA_P(0);
	appendStringInfo(cmsTopnInfoString, "Sketch depth = %d, Sketch width = %d, "
	                 "Size = %ukB", cmsTopn->sketchDepth, cmsTopn->sketchWidth,
	                 VARSIZE(cmsTopn) / 1024);

	PG_RETURN_TEXT_P(CStringGetTextDatum(cmsTopnInfoString->data));
}

/*
 * List a directory (returns the filenames only)
 */
Datum
pg_ls_dir(PG_FUNCTION_ARGS)
{
	FuncCallContext *funcctx;
	struct dirent *de;
	directory_fctx *fctx;

	if (!superuser())
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 (errmsg("must be superuser to get directory listings"))));

	if (SRF_IS_FIRSTCALL())
	{
		MemoryContext oldcontext;

		funcctx = SRF_FIRSTCALL_INIT();
		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);

		fctx = palloc(sizeof(directory_fctx));
		fctx->location = convert_and_check_filename(PG_GETARG_TEXT_P(0));

		fctx->dirdesc = AllocateDir(fctx->location);

		if (!fctx->dirdesc)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not open directory \"%s\": %m",
							fctx->location)));

		funcctx->user_fctx = fctx;
		MemoryContextSwitchTo(oldcontext);
	}

	funcctx = SRF_PERCALL_SETUP();
	fctx = (directory_fctx *) funcctx->user_fctx;

	while ((de = ReadDir(fctx->dirdesc, fctx->location)) != NULL)
	{
		if (strcmp(de->d_name, ".") == 0 ||
			strcmp(de->d_name, "..") == 0)
			continue;

		SRF_RETURN_NEXT(funcctx, CStringGetTextDatum(de->d_name));
	}

	FreeDir(fctx->dirdesc);

	SRF_RETURN_DONE(funcctx);
}

/* AddUpdResqueueCapabilityEntryInternal:
 *
 * Internal function to add a new entry to pg_resqueuecapability, or
 * update an existing one.  Key cols are queueid, restypint.  If
 * old_tuple is set (ie not InvalidOid), the update the ressetting column,
 * else insert a new row.
 *
 */
static void
AddUpdResqueueCapabilityEntryInternal(
								  Relation		 resqueuecap_rel,
								  Oid			 queueid,
								  int			 resTypeInt,
								  char			*pResSetting,
								  Relation		 rel,
								  HeapTuple		 old_tuple)
{
	HeapTuple	new_tuple;
	Datum		values[Natts_pg_resqueuecapability];
	bool		isnull[Natts_pg_resqueuecapability];
	bool		new_record_repl[Natts_pg_resqueuecapability];

	MemSet(isnull, 0, sizeof(bool) * Natts_pg_resqueuecapability);
	MemSet(new_record_repl, 0, sizeof(bool) * Natts_pg_resqueuecapability);

	values[Anum_pg_resqueuecapability_resqueueid - 1]  = 
			ObjectIdGetDatum(queueid);
	values[Anum_pg_resqueuecapability_restypid - 1]   = resTypeInt;

	Assert(pResSetting);

	values[Anum_pg_resqueuecapability_ressetting - 1] = 
			CStringGetTextDatum(pResSetting);
	/* set this column to update */
	new_record_repl[Anum_pg_resqueuecapability_ressetting - 1]  = true;

	ValidateResqueueCapabilityEntry(resTypeInt, pResSetting);

	if (HeapTupleIsValid(old_tuple))
	{
		new_tuple = heap_modify_tuple(old_tuple,
									  RelationGetDescr(resqueuecap_rel),
									  values, isnull, new_record_repl);

		simple_heap_update(resqueuecap_rel, &old_tuple->t_self, new_tuple);
		CatalogUpdateIndexes(resqueuecap_rel, new_tuple);
	}
	else
	{
		new_tuple = heap_form_tuple(RelationGetDescr(resqueuecap_rel), values, isnull);

		simple_heap_insert(resqueuecap_rel, new_tuple);
		CatalogUpdateIndexes(resqueuecap_rel, new_tuple);
	}

	if (HeapTupleIsValid(old_tuple))
		heap_freetuple(new_tuple);
} /* end AddUpdResqueueCapabilityEntryInternal */

/*
 * Helper function
 */
Datum *
create_guc_datum_array(List *guc_list, int num)
{
	Datum      *darray;
	ListCell   *item;
	int         i;

	darray = (Datum *) palloc0(num * sizeof(Datum));

	i = 0;
	foreach(item, guc_list)
		darray[i++] = CStringGetTextDatum((char *) lfirst(item));

	return darray;
}