Datum
pg_database_size_name(PG_FUNCTION_ARGS)
{
	int64		size = 0;
	Name		dbName = PG_GETARG_NAME(0);
	Oid			dbOid = get_database_oid(NameStr(*dbName));

	if (!OidIsValid(dbOid))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_DATABASE),
				 errmsg("database \"%s\" does not exist",
						NameStr(*dbName))));
						
	size = calculate_database_size(dbOid);

	PG_RETURN_INT64(size);
}

Datum
pg_file_write(PG_FUNCTION_ARGS)
{
	FILE	   *f;
	char	   *filename;
	text	   *data;
	int64		count = 0;

	requireSuperuser();

	filename = convert_and_check_filename(PG_GETARG_TEXT_P(0), false);
	data = PG_GETARG_TEXT_P(1);

	if (!PG_GETARG_BOOL(2))
	{
		struct stat fst;

		if (stat(filename, &fst) >= 0)
			ereport(ERROR,
					(ERRCODE_DUPLICATE_FILE,
					 errmsg("file \"%s\" exists", filename)));

		f = fopen(filename, "wb");
	}
	else
		f = fopen(filename, "ab");

	if (!f)
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not open file \"%s\" for writing: %m",
						filename)));

	if (VARSIZE(data) != 0)
	{
		count = fwrite(VARDATA(data), 1, VARSIZE(data) - VARHDRSZ, f);

		if (count != VARSIZE(data) - VARHDRSZ)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not write file \"%s\": %m", filename)));
	}
	fclose(f);

	PG_RETURN_INT64(count);
}

Datum
pg_relation_size(PG_FUNCTION_ARGS)
{
	Oid			relOid = PG_GETARG_OID(0);
	text	   *forkName = PG_GETARG_TEXT_P(1);
	Relation	rel;
	int64		size;

	rel = relation_open(relOid, AccessShareLock);

	size = calculate_relation_size(&(rel->rd_node), rel->rd_backend,
							  forkname_to_number(text_to_cstring(forkName)));

	relation_close(rel, AccessShareLock);

	PG_RETURN_INT64(size);
}

Datum
lo_lseek64(PG_FUNCTION_ARGS)
{
	int32		fd = PG_GETARG_INT32(0);
	int64		offset = PG_GETARG_INT64(1);
	int32		whence = PG_GETARG_INT32(2);
	int64		status;

	if (fd < 0 || fd >= cookies_size || cookies[fd] == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("invalid large-object descriptor: %d", fd)));

	status = inv_seek(cookies[fd], offset, whence);

	PG_RETURN_INT64(status);
}

/*
 * SQL-callable function to clean the insert pending list
 */
Datum
gin_clean_pending_list(PG_FUNCTION_ARGS)
{
	Oid			indexoid = PG_GETARG_OID(0);
	Relation	indexRel = index_open(indexoid, AccessShareLock);
	IndexBulkDeleteResult stats;
	GinState	ginstate;

	if (RecoveryInProgress())
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("recovery is in progress"),
		 errhint("GIN pending list cannot be cleaned up during recovery.")));

	/* Must be a GIN index */
	if (indexRel->rd_rel->relkind != RELKIND_INDEX ||
		indexRel->rd_rel->relam != GIN_AM_OID)
		ereport(ERROR,
				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
				 errmsg("\"%s\" is not a GIN index",
						RelationGetRelationName(indexRel))));

	/*
	 * Reject attempts to read non-local temporary relations; we would be
	 * likely to get wrong data since we have no visibility into the owning
	 * session's local buffers.
	 */
	if (RELATION_IS_OTHER_TEMP(indexRel))
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
			   errmsg("cannot access temporary indexes of other sessions")));

	/* User must own the index (comparable to privileges needed for VACUUM) */
	if (!pg_class_ownercheck(indexoid, GetUserId()))
		aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS,
					   RelationGetRelationName(indexRel));

	memset(&stats, 0, sizeof(stats));
	initGinState(&ginstate, indexRel);
	ginInsertCleanup(&ginstate, true, true, &stats);

	index_close(indexRel, AccessShareLock);

	PG_RETURN_INT64((int64) stats.pages_deleted);
}

/*
 * citus_total_relation_size accepts a table name and returns a distributed table
 * and its indexes' total relation size.
 */
Datum
citus_total_relation_size(PG_FUNCTION_ARGS)
{
	Oid relationId = PG_GETARG_OID(0);
	uint64 totalRelationSize = 0;
	char *tableSizeFunction = PG_TOTAL_RELATION_SIZE_FUNCTION;

	CheckCitusVersion(ERROR);

	if (CStoreTable(relationId))
	{
		tableSizeFunction = CSTORE_TABLE_SIZE_FUNCTION;
	}

	totalRelationSize = DistributedTableSize(relationId, tableSizeFunction);

	PG_RETURN_INT64(totalRelationSize);
}

Datum
pg_file_length(PG_FUNCTION_ARGS)
{
	text	   *filename_t = PG_GETARG_TEXT_P(0);
	char	   *filename;
	struct stat fst;

	requireSuperuser();

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

	PG_RETURN_INT64((int64) fst.st_size);
}

Datum
int48div(PG_FUNCTION_ARGS)
{
    int32		arg1 = PG_GETARG_INT32(0);
    int64		arg2 = PG_GETARG_INT64(1);

    if (arg2 == 0)
    {
        ereport(ERROR,
                (errcode(ERRCODE_DIVISION_BY_ZERO),
                 errmsg("division by zero")));
        /* ensure compiler realizes we mustn't reach the division (gcc bug) */
        PG_RETURN_NULL();
    }

    /* No overflow is possible */
    PG_RETURN_INT64((int64) arg1 / arg2);
}

Datum pgheal_ang2ipix_nest(PG_FUNCTION_ARGS)
{

	q3c_ipix_t nside = PG_GETARG_INT64(0);
	q3c_coord_t ra = PG_GETARG_FLOAT8(1);
	q3c_coord_t dec = PG_GETARG_FLOAT8(2);
	q3c_ipix_t ipix;
	q3c_coord_t theta;
	q3c_coord_t phi;
	if ((!isfinite(ra))||(!isfinite(dec)))
	{
		PG_RETURN_NULL();
	}
	get_theta_phi(ra,dec, &theta, &phi);
	nside_check(nside);
	angle_check(theta);
	ang2pix_nest(nside, theta, phi, &ipix);
	PG_RETURN_INT64(ipix);
}

Datum get_purge_id(PG_FUNCTION_ARGS) {
    VarChar *vis_base64 = PG_GETARG_VARCHAR_P(0);
    visibility_handle_t *vis = deserialize_vis(vis_base64);
    if (!vis) {
        PG_RETURN_BOOL(false);
    }

    char *error = NULL;
    purge_info_t purge_info;
    purge_info.id = -1;
    ezbake_get_purge_info(vis, &purge_info, &error);
    if (error) {
        ereport(ERROR, (errmsg("Error getting the purge info from visibility! %s", error)));
        free(error);
    }

    ezbake_visibility_handle_free(vis);
    PG_RETURN_INT64(purge_info.id);
}

/*
 *	hashgetbitmap() -- get all tuples at once
 */
Datum
hashgetbitmap(PG_FUNCTION_ARGS)
{
	IndexScanDesc scan = (IndexScanDesc) PG_GETARG_POINTER(0);
	TIDBitmap  *tbm = (TIDBitmap *) PG_GETARG_POINTER(1);
	HashScanOpaque so = (HashScanOpaque) scan->opaque;
	bool		res;
	int64		ntids = 0;

	res = _hash_first(scan, ForwardScanDirection);

	while (res)
	{
		bool		add_tuple;

		/*
		 * Skip killed tuples if asked to.
		 */
		if (scan->ignore_killed_tuples)
		{
			Page		page;
			OffsetNumber offnum;

			offnum = ItemPointerGetOffsetNumber(&(so->hashso_curpos));
			page = BufferGetPage(so->hashso_curbuf);
			add_tuple = !ItemIdIsDead(PageGetItemId(page, offnum));
		}
		else
			add_tuple = true;

		/* Save tuple ID, and continue scanning */
		if (add_tuple)
		{
			/* Note we mark the tuple ID as requiring recheck */
			tbm_add_tuples(tbm, &(so->hashso_heappos), 1, true);
			ntids++;
		}

		res = _hash_next(scan, ForwardScanDirection);
	}

	PG_RETURN_INT64(ntids);
}

Datum
int8_dist(PG_FUNCTION_ARGS)
{
	int64		a = PG_GETARG_INT64(0);
	int64		b = PG_GETARG_INT64(1);
	int64		r;
	int64		ra;

	r = a - b;
	ra = Abs(r);

	/* Overflow check. */
	if (ra < 0 || (!SAMESIGN(a, b) && !SAMESIGN(r, a)))
		ereport(ERROR,
				(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
				 errmsg("bigint out of range")));

	PG_RETURN_INT64(ra);
}

Datum
pg_stat_get_db_conflict_all(PG_FUNCTION_ARGS)
{
	Oid			dbid = PG_GETARG_OID(0);
	int64		result;
	PgStat_StatDBEntry *dbentry;

	if ((dbentry = pgstat_fetch_stat_dbentry(dbid)) == NULL)
		result = 0;
	else
		result = (int64) (
						  dbentry->n_conflict_tablespace +
						  dbentry->n_conflict_lock +
						  dbentry->n_conflict_snapshot +
						  dbentry->n_conflict_bufferpin +
						  dbentry->n_conflict_startup_deadlock);

	PG_RETURN_INT64(result);
}

Datum
int8inc(PG_FUNCTION_ARGS)
{
    /*
     * When int8 is pass-by-reference, we provide this special case to avoid
     * palloc overhead for COUNT(): when called as an aggregate, we know that
     * the argument is modifiable local storage, so just update it in-place.
     * (If int8 is pass-by-value, then of course this is useless as well as
     * incorrect, so just ifdef it out.)
     */
#ifndef USE_FLOAT8_BYVAL		/* controls int8 too */
    if (AggCheckCallContext(fcinfo, NULL))
    {
        int64	   *arg = (int64 *) PG_GETARG_POINTER(0);
        int64		result;

        result = *arg + 1;
        /* Overflow check */
        if (result < 0 && *arg > 0)
            ereport(ERROR,
                    (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
                     errmsg("bigint out of range")));

        *arg = result;
        PG_RETURN_POINTER(arg);
    }
    else
#endif
    {
        /* Not called as an aggregate, so just do it the dumb way */
        int64		arg = PG_GETARG_INT64(0);
        int64		result;

        result = arg + 1;
        /* Overflow check */
        if (result < 0 && arg > 0)
            ereport(ERROR,
                    (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
                     errmsg("bigint out of range")));

        PG_RETURN_INT64(result);
    }
}

Datum
pg_stat_get_xact_tuples_deleted(PG_FUNCTION_ARGS)
{
	Oid			relid = PG_GETARG_OID(0);
	int64		result;
	PgStat_TableStatus *tabentry;
	PgStat_TableXactStatus *trans;

	if ((tabentry = find_tabstat_entry(relid)) == NULL)
		result = 0;
	else
	{
		result = tabentry->t_counts.t_tuples_deleted;
		/* live subtransactions' counts aren't in t_tuples_deleted yet */
		for (trans = tabentry->trans; trans != NULL; trans = trans->upper)
			result += trans->tuples_deleted;
	}

	PG_RETURN_INT64(result);
}

Datum
int82pl(PG_FUNCTION_ARGS)
{
    int64		arg1 = PG_GETARG_INT64(0);
    int16		arg2 = PG_GETARG_INT16(1);
    int64		result;

    result = arg1 + arg2;

    /*
     * Overflow check.	If the inputs are of different signs then their sum
     * cannot overflow.  If the inputs are of the same sign, their sum had
     * better be that sign too.
     */
    if (SAMESIGN(arg1, arg2) && !SAMESIGN(result, arg1))
        ereport(ERROR,
                (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
                 errmsg("bigint out of range")));
    PG_RETURN_INT64(result);
}

Datum
relation_size(PG_FUNCTION_ARGS)
{
	text	   *relname = PG_GETARG_TEXT_P(0);
	RangeVar   *relrv;
	Relation	relation;
	Oid			relnodeOid;
	Oid         tblspcOid;

	relrv = makeRangeVarFromNameList(textToQualifiedNameList(relname,
													   "relation_size"));
	relation = relation_openrv(relrv, AccessShareLock);

	tblspcOid  = relation->rd_rel->reltablespace;
	relnodeOid = relation->rd_rel->relfilenode;

	relation_close(relation, AccessShareLock);

	PG_RETURN_INT64(calculate_relation_size(tblspcOid, relnodeOid));
}

Datum
pg_database_size_oid(PG_FUNCTION_ARGS)
{
	int64		size = 0;
	Oid			dbOid = PG_GETARG_OID(0);
	size = calculate_database_size(dbOid);
	
	if (Gp_role == GP_ROLE_DISPATCH)
	{
		StringInfoData buffer;
		
		initStringInfo(&buffer);

		appendStringInfo(&buffer, "select sum(pg_database_size(%u))::int8 from gp_dist_random('gp_id');", dbOid);

		size += get_size_from_segDBs(buffer.data);
	}

	PG_RETURN_INT64(size);
}

Datum
int82mi(PG_FUNCTION_ARGS)
{
    int64		arg1 = PG_GETARG_INT64(0);
    int16		arg2 = PG_GETARG_INT16(1);
    int64		result;

    result = arg1 - arg2;

    /*
     * Overflow check.	If the inputs are of the same sign then their
     * difference cannot overflow.	If they are of different signs then the
     * result should be of the same sign as the first input.
     */
    if (!SAMESIGN(arg1, arg2) && !SAMESIGN(result, arg1))
        ereport(ERROR,
                (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
                 errmsg("bigint out of range")));
    PG_RETURN_INT64(result);
}

/*
 * txid_current() returns int8
 *
 *	Return the current toplevel transaction ID as TXID
 *	If the current transaction does not have one, one is assigned.
 */
Datum
txid_current(PG_FUNCTION_ARGS)
{
	txid		val;
	TxidEpoch	state;

	/*
	 * Must prevent during recovery because if an xid is not assigned we try
	 * to assign one, which would fail. Programs already rely on this function
	 * to always return a valid current xid, so we should not change this to
	 * return NULL or similar invalid xid.
	 */
	PreventCommandDuringRecovery("txid_current()");

	load_xid_epoch(&state);

	val = convert_xid(GetTopTransactionId(), &state);

	PG_RETURN_INT64(val);
}