/* cash_numeric()
 * Convert cash to numeric.
 */
Datum
cash_numeric(PG_FUNCTION_ARGS)
{
	Cash		money = PG_GETARG_CASH(0);
	Numeric		result;
	int			fpoint;
	int64		scale;
	int			i;
	Datum		amount;
	Datum		numeric_scale;
	Datum		quotient;
	struct lconv *lconvert = PGLC_localeconv();

	/* see comments about frac_digits in cash_in() */
	fpoint = lconvert->frac_digits;
	if (fpoint < 0 || fpoint > 10)
		fpoint = 2;

	/* compute required scale factor */
	scale = 1;
	for (i = 0; i < fpoint; i++)
		scale *= 10;

	/* form the result as money / scale */
	amount = DirectFunctionCall1(int8_numeric, Int64GetDatum(money));
	numeric_scale = DirectFunctionCall1(int8_numeric, Int64GetDatum(scale));
	quotient = DirectFunctionCall2(numeric_div, amount, numeric_scale);

	/* forcibly round to exactly the intended number of digits */
	result = DatumGetNumeric(DirectFunctionCall2(numeric_round,
												 quotient,
												 Int32GetDatum(fpoint)));

	PG_RETURN_NUMERIC(result);
}

void GpPersistentDatabaseNode_SetDatumValues(
	Datum							*values,

	Oid 							tablespaceOid,
	Oid 							databaseOid,
	PersistentFileSysState			persistentState,
	int64							createMirrorDataLossTrackingSessionNum,
	MirroredObjectExistenceState	mirrorExistenceState,
	int32							reserved,
	TransactionId					parentXid,
	int64							persistentSerialNum)
{
	values[Anum_gp_persistent_database_node_tablespace_oid - 1] = 
									ObjectIdGetDatum(tablespaceOid);
	values[Anum_gp_persistent_database_node_database_oid - 1] = 
									ObjectIdGetDatum(databaseOid);

	values[Anum_gp_persistent_database_node_persistent_state - 1] = 
									Int16GetDatum(persistentState);

	values[Anum_gp_persistent_database_node_create_mirror_data_loss_tracking_session_num - 1] = 
									Int64GetDatum(createMirrorDataLossTrackingSessionNum);

	values[Anum_gp_persistent_database_node_mirror_existence_state - 1] = 
									Int16GetDatum(mirrorExistenceState);

	values[Anum_gp_persistent_database_node_reserved - 1] = 
									Int32GetDatum(reserved);

	values[Anum_gp_persistent_database_node_parent_xid - 1] = 
									Int32GetDatum(parentXid);

	values[Anum_gp_persistent_database_node_persistent_serial_num - 1] = 
									Int64GetDatum(persistentSerialNum);
}

/* int8_cash()
 * Convert int8 (bigint) to cash
 */
Datum
int8_cash(PG_FUNCTION_ARGS)
{
	int64		amount = PG_GETARG_INT64(0);
	Cash		result;
	int			fpoint;
	int64		scale;
	int			i;
	struct lconv *lconvert = PGLC_localeconv();

	/* see comments about frac_digits in cash_in() */
	fpoint = lconvert->frac_digits;
	if (fpoint < 0 || fpoint > 10)
		fpoint = 2;

	/* compute required scale factor */
	scale = 1;
	for (i = 0; i < fpoint; i++)
		scale *= 10;

	/* compute amount * scale, checking for overflow */
	result = DatumGetInt64(DirectFunctionCall2(int8mul, Int64GetDatum(amount),
											   Int64GetDatum(scale)));

	PG_RETURN_CASH(result);
}

void GpRelationNode_SetDatumValues(
	Datum							*values,
	Oid 							tablespaceOid,
	Oid 							relfilenodeOid,
	int32							segmentFileNum,
	int64							createMirrorDataLossTrackingSessionNum,
	ItemPointer		 				persistentTid,
	int64							persistentSerialNum)
{
	values[Anum_gp_relation_node_tablespace_oid - 1] =
		ObjectIdGetDatum(tablespaceOid);

	values[Anum_gp_relation_node_relfilenode_oid - 1] = 
									ObjectIdGetDatum(relfilenodeOid);

	values[Anum_gp_relation_node_segment_file_num - 1] = 
									Int32GetDatum(segmentFileNum);

	values[Anum_gp_relation_node_create_mirror_data_loss_tracking_session_num - 1] = 
									Int64GetDatum(createMirrorDataLossTrackingSessionNum);

	values[Anum_gp_relation_node_persistent_tid - 1] =
									PointerGetDatum(persistentTid);
	
	values[Anum_gp_relation_node_persistent_serial_num - 1] = 
									Int64GetDatum(persistentSerialNum);
}

Datum
pgsysconf(PG_FUNCTION_ARGS)
{
	HeapTuple	tuple;
	TupleDesc	tupdesc;
	Datum		values[PGSYSCONF_COLS];
	bool		nulls[PGSYSCONF_COLS];

	/* initialize nulls array to build the tuple */
	memset(nulls, 0, sizeof(nulls));

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

	/* Page size */
	values[0] = Int64GetDatum(sysconf(_SC_PAGESIZE));

	/* free page in memory */
	values[1] = Int64GetDatum(sysconf(_SC_AVPHYS_PAGES));

	/* total memory */
	values[2] = Int64GetDatum(sysconf(_SC_PHYS_PAGES));

	/* Build and return the result tuple. */
	tuple = heap_form_tuple(tupdesc, values, nulls);
	PG_RETURN_DATUM( HeapTupleGetDatum(tuple) );
}

/*
 * Compute the difference in bytes between two WAL locations.
 */
Datum
pg_xlog_location_diff(PG_FUNCTION_ARGS)
{
	text	   *location1 = PG_GETARG_TEXT_P(0);
	text	   *location2 = PG_GETARG_TEXT_P(1);
	char	   *str1,
			   *str2;
	XLogRecPtr	loc1,
				loc2;
	Numeric		result;

	/*
	 * Read and parse input
	 */
	str1 = text_to_cstring(location1);
	str2 = text_to_cstring(location2);

	validate_xlog_location(str1);
	validate_xlog_location(str2);

	if (sscanf(str1, "%X/%X", &loc1.xlogid, &loc1.xrecoff) != 2)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
		   errmsg("could not parse transaction log location \"%s\"", str1)));
	if (sscanf(str2, "%X/%X", &loc2.xlogid, &loc2.xrecoff) != 2)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
		   errmsg("could not parse transaction log location \"%s\"", str2)));

	/*
	 * Sanity check
	 */
	if (loc1.xrecoff > XLogFileSize)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("xrecoff \"%X\" is out of valid range, 0..%X", loc1.xrecoff, XLogFileSize)));
	if (loc2.xrecoff > XLogFileSize)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("xrecoff \"%X\" is out of valid range, 0..%X", loc2.xrecoff, XLogFileSize)));

	/*
	 * result = XLogFileSize * (xlogid1 - xlogid2) + xrecoff1 - xrecoff2
	 */
	result = DatumGetNumeric(DirectFunctionCall2(numeric_sub,
	   DirectFunctionCall1(int8_numeric, Int64GetDatum((int64) loc1.xlogid)),
	 DirectFunctionCall1(int8_numeric, Int64GetDatum((int64) loc2.xlogid))));
	result = DatumGetNumeric(DirectFunctionCall2(numeric_mul,
	  DirectFunctionCall1(int8_numeric, Int64GetDatum((int64) XLogFileSize)),
												 NumericGetDatum(result)));
	result = DatumGetNumeric(DirectFunctionCall2(numeric_add,
												 NumericGetDatum(result),
	DirectFunctionCall1(int8_numeric, Int64GetDatum((int64) loc1.xrecoff))));
	result = DatumGetNumeric(DirectFunctionCall2(numeric_sub,
												 NumericGetDatum(result),
	DirectFunctionCall1(int8_numeric, Int64GetDatum((int64) loc2.xrecoff))));

	PG_RETURN_NUMERIC(result);
}

/*
 * Count the number of all-visible and all-frozen pages in the visibility
 * map for a particular relation.
 */
Datum
pg_visibility_map_summary(PG_FUNCTION_ARGS)
{
	Oid			relid = PG_GETARG_OID(0);
	Relation	rel;
	BlockNumber nblocks;
	BlockNumber blkno;
	Buffer		vmbuffer = InvalidBuffer;
	int64		all_visible = 0;
	int64		all_frozen = 0;
	TupleDesc	tupdesc;
	Datum		values[2];
	bool		nulls[2];

	rel = relation_open(relid, AccessShareLock);

	/* Only some relkinds have a visibility map */
	check_relation_relkind(rel);

	nblocks = RelationGetNumberOfBlocks(rel);

	for (blkno = 0; blkno < nblocks; ++blkno)
	{
		int32		mapbits;

		/* Make sure we are interruptible. */
		CHECK_FOR_INTERRUPTS();

		/* Get map info. */
		mapbits = (int32) visibilitymap_get_status(rel, blkno, &vmbuffer);
		if ((mapbits & VISIBILITYMAP_ALL_VISIBLE) != 0)
			++all_visible;
		if ((mapbits & VISIBILITYMAP_ALL_FROZEN) != 0)
			++all_frozen;
	}

	/* Clean up. */
	if (vmbuffer != InvalidBuffer)
		ReleaseBuffer(vmbuffer);
	relation_close(rel, AccessShareLock);

	tupdesc = CreateTemplateTupleDesc(2, false);
	TupleDescInitEntry(tupdesc, (AttrNumber) 1, "all_visible", INT8OID, -1, 0);
	TupleDescInitEntry(tupdesc, (AttrNumber) 2, "all_frozen", INT8OID, -1, 0);
	tupdesc = BlessTupleDesc(tupdesc);

	MemSet(nulls, 0, sizeof(nulls));
	values[0] = Int64GetDatum(all_visible);
	values[1] = Int64GetDatum(all_frozen);

	PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
}

Datum HASHAPI_Hash_1_BigInt(PG_FUNCTION_ARGS)
{
    int32 num_segs;    /* number of segments  */
	int16 algorithm;  /* hashing algorithm   */
	int64 val1;        /* big int input value */
    unsigned int targetbucket; /* 0-based  */
	Datum d1;
	Oid oid;

	/* Get number of segments */
    num_segs = PG_GETARG_INT32(0);
	
	/* Get hashing algoriithm */
	algorithm = PG_GETARG_INT16(1);
    
	/* Get the value to hash */
	val1 = PG_GETARG_INT64(2);
	
	d1 = Int64GetDatum(val1);
	
	/* create a CdbHash for this hash test. */
    h = makeCdbHash(num_segs, algorithm);
	
	/* init cdb hash */
	cdbhashinit(h);
	oid = INT8OID;
	cdbhash(h, d1, oid);
	
	/* reduce the result hash value */
	targetbucket = cdbhashreduce(h);	
	
    PG_RETURN_INT32(targetbucket); /* return target bucket (segID) */
}

/*
 * load_shard_id_array returns the shard identifiers for a particular
 * distributed table as a bigint array. Uses pg_shard's shard interval
 * cache if the second parameter is true, otherwise eagerly loads the
 * shard intervals from the backing table.
 */
Datum
load_shard_id_array(PG_FUNCTION_ARGS)
{
	Oid distributedTableId = PG_GETARG_OID(0);
	bool useCache = PG_GETARG_BOOL(1);
	ArrayType *shardIdArrayType = NULL;
	ListCell *shardCell = NULL;
	int shardIdIndex = 0;
	Oid shardIdTypeId = INT8OID;

	List *shardList = NIL;
	int shardIdCount = -1;
	Datum *shardIdDatumArray = NULL;

	if (useCache)
	{
		shardList = LookupShardIntervalList(distributedTableId);
	}
	else
	{
		shardList = LoadShardIntervalList(distributedTableId);
	}

	shardIdCount = list_length(shardList);
	shardIdDatumArray = palloc0(shardIdCount * sizeof(Datum));

	foreach(shardCell, shardList)
	{
		ShardInterval *shardId = (ShardInterval *) lfirst(shardCell);
		Datum shardIdDatum = Int64GetDatum(shardId->id);

		shardIdDatumArray[shardIdIndex] = shardIdDatum;
		shardIdIndex++;
	}

/*
 * PrunedShardIdsForTable loads the shard intervals for the specified table,
 * prunes them using the provided clauses. It returns an ArrayType containing
 * the shard identifiers, suitable for return from an SQL-facing function.
 */
static ArrayType *
PrunedShardIdsForTable(Oid distributedTableId, List *whereClauseList)
{
	ArrayType *shardIdArrayType = NULL;
	ListCell *shardCell = NULL;
	int shardIdIndex = 0;
	Oid shardIdTypeId = INT8OID;

	List *shardList = LoadShardIntervalList(distributedTableId);
	int shardIdCount = -1;
	Datum *shardIdDatumArray = NULL;

	shardList = PruneShardList(distributedTableId, whereClauseList, shardList);

	shardIdCount = list_length(shardList);
	shardIdDatumArray = palloc0(shardIdCount * sizeof(Datum));

	foreach(shardCell, shardList)
	{
		ShardInterval *shardId = (ShardInterval *) lfirst(shardCell);
		Datum shardIdDatum = Int64GetDatum(shardId->id);

		shardIdDatumArray[shardIdIndex] = shardIdDatum;
		shardIdIndex++;
	}

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

/* numeric_cash()
 * Convert numeric to cash.
 */
Datum
numeric_cash(PG_FUNCTION_ARGS)
{
	Datum		amount = PG_GETARG_DATUM(0);
	Cash		result;
	int			fpoint;
	int64		scale;
	int			i;
	Datum		numeric_scale;
	struct lconv *lconvert = PGLC_localeconv();

	/* see comments about frac_digits in cash_in() */
	fpoint = lconvert->frac_digits;
	if (fpoint < 0 || fpoint > 10)
		fpoint = 2;

	/* compute required scale factor */
	scale = 1;
	for (i = 0; i < fpoint; i++)
		scale *= 10;

	/* multiply the input amount by scale factor */
	numeric_scale = DirectFunctionCall1(int8_numeric, Int64GetDatum(scale));
	amount = DirectFunctionCall2(numeric_mul, amount, numeric_scale);

	/* note that numeric_int8 will round to nearest integer for us */
	result = DatumGetInt64(DirectFunctionCall1(numeric_int8, amount));

	PG_RETURN_CASH(result);
}

/*
 * LoadShardAlias finds the row for given relation and shardId in pg_dist_shard,
 * finds the shard alias in this row if any, and then deep copies this alias.
 */
char *
LoadShardAlias(Oid relationId, uint64 shardId)
{
	SysScanDesc scanDescriptor = NULL;
	ScanKeyData scanKey[1];
	int scanKeyCount = 1;
	HeapTuple heapTuple = NULL;
	Datum shardAliasDatum = 0;
	bool shardAliasNull = false;
	char *shardAlias = NULL;

	Relation pgDistShard = heap_open(DistShardRelationId(), AccessShareLock);
	TupleDesc tupleDescriptor = RelationGetDescr(pgDistShard);

	ScanKeyInit(&scanKey[0], Anum_pg_dist_shard_shardid,
				BTEqualStrategyNumber, F_INT8EQ, Int64GetDatum(shardId));

	scanDescriptor = systable_beginscan(pgDistShard,
										DistShardShardidIndexId(), true,
										NULL, scanKeyCount, scanKey);

	/*
	 * Normally, we should have at most one tuple here as we have a unique index
	 * on shardId. However, if users want to drop this uniqueness constraint,
	 * and look up the shardalias based on the relation and shardId pair, we
	 * still allow that. We don't have any users relaying on this feature. Thus,
	 * we may consider to remove this check.
	 */
	heapTuple = systable_getnext(scanDescriptor);
	while (HeapTupleIsValid(heapTuple))
	{
		Form_pg_dist_shard pgDistShardForm = (Form_pg_dist_shard) GETSTRUCT(heapTuple);
		if (pgDistShardForm->logicalrelid == relationId)
		{
			break;
		}

		heapTuple = systable_getnext(scanDescriptor);
	}

	/* if no tuple found, error out */
	if (!HeapTupleIsValid(heapTuple))
	{
		ereport(ERROR, (errmsg("could not find valid entry for relationId: %u "
							   "and shard " UINT64_FORMAT, relationId, shardId)));
	}

	/* if shard alias exists, deep copy cstring */
	shardAliasDatum = heap_getattr(heapTuple, Anum_pg_dist_shard_shardalias,
								   tupleDescriptor, &shardAliasNull);
	if (!shardAliasNull)
	{
		shardAlias = TextDatumGetCString(shardAliasDatum);
	}

	systable_endscan(scanDescriptor);
	heap_close(pgDistShard, AccessShareLock);

	return shardAlias;
}

/*
 * gp_workfile_mgr_reset_segspace
 *    Function to reset the used segspace on a segment
 *    This directly manipulates the segspace counter and
 *    should be used for testing purposes only
 *  Returns the size before the reset
 */
Datum
gp_workfile_mgr_reset_segspace(PG_FUNCTION_ARGS)
{
	int64 size = WorkfileSegspace_GetSize();
	WorkfileSegspace_Commit(0, size);
	return Int64GetDatum(size);
}

void GpPersistentTablespaceNode_SetDatumValues(
	Datum							*values,

	Oid 							filespaceOid,
	Oid 							tablespaceOid,
	PersistentFileSysState			persistentState,
	TransactionId					parentXid,
	int64							persistentSerialNum,
	ItemPointerData 				*previousFreeTid,
	bool							sharedStorage)
{

	values[Anum_gp_persistent_tablespace_node_filespace_oid - 1] = 
									ObjectIdGetDatum(filespaceOid);

	values[Anum_gp_persistent_tablespace_node_tablespace_oid - 1] = 
									ObjectIdGetDatum(tablespaceOid);

	values[Anum_gp_persistent_tablespace_node_persistent_state - 1] = 
									Int16GetDatum(persistentState);

	values[Anum_gp_persistent_tablespace_node_reserved - 1] = 
									Int32GetDatum(0);

	values[Anum_gp_persistent_tablespace_node_parent_xid - 1] = 
									Int32GetDatum(parentXid);

	values[Anum_gp_persistent_tablespace_node_persistent_serial_num - 1] = 
									Int64GetDatum(persistentSerialNum);

	values[Anum_gp_persistent_tablespace_node_previous_free_tid - 1] =
									PointerGetDatum(previousFreeTid);
}

/*
 * Returns the number of bytes used for workfiles on a segment
 * according to WorkfileDiskspace
 */
Datum
gp_workfile_mgr_used_diskspace(PG_FUNCTION_ARGS)
{
	/*
	 * Build a tuple descriptor for our result type
	 * The number and type of attributes have to match the definition of the
	 * view gp_workfile_mgr_diskspace
	 */
	TupleDesc tupdesc = CreateTemplateTupleDesc(NUM_USED_DISKSPACE_ELEM, false);

	TupleDescInitEntry(tupdesc, (AttrNumber) 1, "segid",
			INT4OID, -1 /* typmod */, 0 /* attdim */);
	TupleDescInitEntry(tupdesc, (AttrNumber) 2, "bytes",
			INT8OID, -1 /* typmod */, 0 /* attdim */);

	tupdesc =  BlessTupleDesc(tupdesc);

	Datum		values[NUM_USED_DISKSPACE_ELEM];
	bool		nulls[NUM_USED_DISKSPACE_ELEM];
	MemSet(nulls, 0, sizeof(nulls));

	values[0] = Int32GetDatum(GpIdentity.segindex);
	values[1] = Int64GetDatum(WorkfileSegspace_GetSize());

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

	PG_RETURN_DATUM(result);
}

/*
 * txid_snapshot_xip(txid_snapshot) returns setof int8
 *
 *		return in-progress TXIDs in snapshot.
 */
Datum
txid_snapshot_xip(PG_FUNCTION_ARGS)
{
	FuncCallContext *fctx;
	TxidSnapshot *snap;
	txid		value;

	/* on first call initialize snap_state and get copy of snapshot */
	if (SRF_IS_FIRSTCALL())
	{
		TxidSnapshot *arg = (TxidSnapshot *) PG_GETARG_VARLENA_P(0);

		fctx = SRF_FIRSTCALL_INIT();

		/* make a copy of user snapshot */
		snap = MemoryContextAlloc(fctx->multi_call_memory_ctx, VARSIZE(arg));
		memcpy(snap, arg, VARSIZE(arg));

		fctx->user_fctx = snap;
	}

	/* return values one-by-one */
	fctx = SRF_PERCALL_SETUP();
	snap = fctx->user_fctx;
	if (fctx->call_cntr < snap->nxip)
	{
		value = snap->xip[fctx->call_cntr];
		SRF_RETURN_NEXT(fctx, Int64GetDatum(value));
	}
	else
	{
		SRF_RETURN_DONE(fctx);
	}
}

static Datum
leftmostvalue_money(void)
{
	/*
	 * Use sequence's definition to keep compatibility.
	 */
	return Int64GetDatum(SEQ_MINVALUE);
}

static HeapTuple
FormErrorTuple(CdbSreh *cdbsreh)
{
	bool		nulls[NUM_ERRORTABLE_ATTR];
	Datum		values[NUM_ERRORTABLE_ATTR];
	MemoryContext oldcontext;
					
	oldcontext = MemoryContextSwitchTo(cdbsreh->badrowcontext);
	
	/* Initialize all values for row to NULL */
	MemSet(values, 0, NUM_ERRORTABLE_ATTR * sizeof(Datum));
	MemSet(nulls, true, NUM_ERRORTABLE_ATTR * sizeof(bool));
	
	/* command start time */
	values[errtable_cmdtime - 1] = TimestampTzGetDatum(GetCurrentStatementStartTimestamp());
	nulls[errtable_cmdtime - 1] = false;
		
	/* line number */
	if (cdbsreh->linenumber > 0)
	{
		values[errtable_linenum - 1] = Int64GetDatum(cdbsreh->linenumber);
		nulls[errtable_linenum - 1] = false;
	}

	if(cdbsreh->is_server_enc)
	{
		/* raw data */
		values[errtable_rawdata - 1] = DirectFunctionCall1(textin, CStringGetDatum(cdbsreh->rawdata));
		nulls[errtable_rawdata - 1] = false;
	}
	else
	{
		/* raw bytes */
		PreprocessByteaData(cdbsreh->rawdata);
		values[errtable_rawbytes - 1] = DirectFunctionCall1(byteain, CStringGetDatum(cdbsreh->rawdata));
		nulls[errtable_rawbytes - 1] = false;
	}

	/* file name */
	values[errtable_filename - 1] = DirectFunctionCall1(textin, CStringGetDatum(cdbsreh->filename));
	nulls[errtable_filename - 1] = false;

	/* relation name */
	values[errtable_relname - 1] = DirectFunctionCall1(textin, CStringGetDatum(cdbsreh->relname));
	nulls[errtable_relname - 1] = false;
	
	/* error message */
	values[errtable_errmsg - 1] = DirectFunctionCall1(textin, CStringGetDatum(cdbsreh->errmsg));
	nulls[errtable_errmsg - 1] = false;
	
	
	MemoryContextSwitchTo(oldcontext);
	
	/*
	 * And now we can form the input tuple.
	 */
	return heap_form_tuple(GetErrorTupleDesc(), values, nulls);
}

/*
 * GetFastSequences
 *
 * Get a list of consecutive sequence numbers. The starting sequence
 * number is the maximal value between 'lastsequence' + 1 and minSequence.
 * The length of the list is given.
 *
 * If there is not such an entry for objid in the table, create
 * one here.
 *
 * The existing entry for objid in the table is updated with a new
 * lastsequence value.
 */
int64 GetFastSequences(Oid objid, int64 objmod,
					   int64 minSequence, int64 numSequences)
{
	Relation gp_fastsequence_rel;
	TupleDesc tupleDesc;
	HeapTuple tuple;
	cqContext	 cqc;
	int64 firstSequence = minSequence;
	Datum lastSequenceDatum;
	int64 newLastSequence;

	gp_fastsequence_rel = heap_open(FastSequenceRelationId, RowExclusiveLock);
	tupleDesc = RelationGetDescr(gp_fastsequence_rel);
	
	tuple = caql_getfirst(
			caql_addrel(cqclr(&cqc), gp_fastsequence_rel),
			cql("SELECT * FROM gp_fastsequence "
				" WHERE objid = :1 "
				" AND objmod = :2 "
				" FOR UPDATE ",
				ObjectIdGetDatum(objid),
				Int64GetDatum(objmod)));

	if (!HeapTupleIsValid(tuple))
	{
		newLastSequence = firstSequence + numSequences - 1;
	}
	else
	{
		bool isNull;

		lastSequenceDatum = heap_getattr(tuple, Anum_gp_fastsequence_last_sequence,
										tupleDesc, &isNull);
		
		if (isNull)
			ereport(ERROR,
					(errcode(ERRCODE_UNDEFINED_OBJECT),
					 errmsg("got an invalid lastsequence number: NULL")));
		
		if (DatumGetInt64(lastSequenceDatum) + 1 > firstSequence)
			firstSequence = DatumGetInt64(lastSequenceDatum) + 1;
		newLastSequence = firstSequence + numSequences - 1;
	}
	
	update_fastsequence(gp_fastsequence_rel, tuple, tupleDesc,
						objid, objmod, newLastSequence);

	if (HeapTupleIsValid(tuple))
	{
		heap_freetuple(tuple);
	}
		
	/* Refer to the comment at the end of InsertFastSequenceEntry. */
	heap_close(gp_fastsequence_rel, RowExclusiveLock);

	return firstSequence;
}