/*
 * text_regclass: convert text to regclass
 *
 * This could be replaced by CoerceViaIO, except that we need to treat
 * text-to-regclass as an implicit cast to support legacy forms of nextval()
 * and related functions.
 */
Datum
text_regclass(PG_FUNCTION_ARGS)
{
	text	   *relname = PG_GETARG_TEXT_P(0);
	Oid			result;
	RangeVar   *rv;

	rv = makeRangeVarFromNameList(textToQualifiedNameList(relname));

	/* We might not even have permissions on this relation; don't lock it. */
	result = RangeVarGetRelid(rv, NoLock, false);

	PG_RETURN_OID(result);
}

/*
 * lo_import -
 *	  imports a file as an (inversion) large object.
 */
Datum
lo_import(PG_FUNCTION_ARGS)
{
	text	   *filename = PG_GETARG_TEXT_PP(0);

#ifdef PGXC
	ereport(ERROR,
			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
			 errmsg("Postgres-XC does not support large object yet"),
			 errdetail("The feature is not currently supported")));
#endif

	PG_RETURN_OID(lo_import_internal(filename, InvalidOid));
}

Datum
lo_create(PG_FUNCTION_ARGS)
{
	Oid			lobjId = PG_GETARG_OID(0);

	/*
	 * We don't actually need to store into fscxt, but create it anyway to
	 * ensure that AtEOXact_LargeObject knows there is state to clean up
	 */
	CreateFSContext();

	lobjId = inv_create(lobjId);

	PG_RETURN_OID(lobjId);
}

Datum
i8tooid(PG_FUNCTION_ARGS)
{
    int64		arg = PG_GETARG_INT64(0);
    Oid			result;

    result = (Oid) arg;

    /* Test for overflow by reverse-conversion. */
    if ((int64) result != arg)
        ereport(ERROR,
                (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
                 errmsg("OID out of range")));

    PG_RETURN_OID(result);
}

/*
 * to_regtype		- converts "typename" to type OID
 *
 * If the name is not found, we return NULL.
 */
Datum
to_regtype(PG_FUNCTION_ARGS)
{
	char	   *typ_name = PG_GETARG_CSTRING(0);
	Oid			result;
	int32		typmod;

	/*
	 * Invoke the full parser to deal with special cases such as array syntax.
	 */
	parseTypeString(typ_name, &result, &typmod, true);

	if (OidIsValid(result))
		PG_RETURN_OID(result);
	else
		PG_RETURN_NULL();
}

/*
 * Create LO with initial contents given by a bytea argument
 */
Datum
lo_from_bytea(PG_FUNCTION_ARGS)
{
	Oid			loOid = PG_GETARG_OID(0);
	bytea	   *str = PG_GETARG_BYTEA_PP(1);
	LargeObjectDesc *loDesc;
	int written PG_USED_FOR_ASSERTS_ONLY;

	CreateFSContext();

	loOid = inv_create(loOid);
	loDesc = inv_open(loOid, INV_WRITE, fscxt);
	written = inv_write(loDesc, VARDATA_ANY(str), VARSIZE_ANY_EXHDR(str));
	Assert(written == VARSIZE_ANY_EXHDR(str));
	inv_close(loDesc);

	PG_RETURN_OID(loOid);
}

Datum
pg_highest_oid(PG_FUNCTION_ARGS __attribute__((unused)))
{
	Oid result;
	Oid max_from_segdbs;

	result = ShmemVariableCache->nextOid;

	if (Gp_role == GP_ROLE_DISPATCH)
	{
		max_from_segdbs = get_max_oid_from_segDBs();

		if (max_from_segdbs > result)
			result = max_from_segdbs;
	}

	PG_RETURN_OID(result);
}

Datum
lo_create(PG_FUNCTION_ARGS)
{
	Oid			lobjId = PG_GETARG_OID(0);

	ereport(ERROR,
		(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
		 errmsg("large objects are not supported")));

	/*
	 * We don't actually need to store into fscxt, but create it anyway to
	 * ensure that AtEOXact_LargeObject knows there is state to clean up
	 */
	CreateFSContext();

	lobjId = inv_create(lobjId);

	PG_RETURN_OID(lobjId);
}

/*
 * to_regproc	- converts "proname" to proc OID
 *
 * If the name is not found, we return NULL.
 */
Datum
to_regproc(PG_FUNCTION_ARGS)
{
	char	   *pro_name = PG_GETARG_CSTRING(0);
	List	   *names;
	FuncCandidateList clist;

	/*
	 * Parse the name into components and see if it matches any pg_proc
	 * entries in the current search path.
	 */
	names = stringToQualifiedNameList(pro_name);
	clist = FuncnameGetCandidates(names, -1, NIL, false, false, true);

	if (clist == NULL || clist->next != NULL)
		PG_RETURN_NULL();

	PG_RETURN_OID(clist->oid);
}

/*
 * to_regoper		- converts "oprname" to operator OID
 *
 * If the name is not found, we return NULL.
 */
Datum
to_regoper(PG_FUNCTION_ARGS)
{
	char	   *opr_name = PG_GETARG_CSTRING(0);
	List	   *names;
	FuncCandidateList clist;

	/*
	 * Parse the name into components and see if it matches any pg_operator
	 * entries in the current search path.
	 */
	names = stringToQualifiedNameList(opr_name);
	clist = OpernameGetCandidates(names, '\0', true);

	if (clist == NULL || clist->next != NULL)
		PG_RETURN_NULL();

	PG_RETURN_OID(clist->oid);
}

Datum
text_oid(PG_FUNCTION_ARGS)
{
	text	   *string = PG_GETARG_TEXT_P(0);
	Oid			result;
	int			len;
	char	   *str;

	len = (VARSIZE(string) - VARHDRSZ);

	str = palloc(len + 1);
	memcpy(str, VARDATA(string), len);
	*(str + len) = '\0';

	result = oidin_subr("text_oid", str, NULL);

	pfree(str);

	PG_RETURN_OID(result);
}

/* Binary I/O support */
Datum
enum_recv(PG_FUNCTION_ARGS)
{
	StringInfo	buf = (StringInfo) PG_GETARG_POINTER(0);
	Oid			enumtypoid = PG_GETARG_OID(1);
	Oid			enumoid;
	HeapTuple	tup;
	char	   *name;
	int			nbytes;

	name = pq_getmsgtext(buf, buf->len - buf->cursor, &nbytes);

	/* must check length to prevent Assert failure within SearchSysCache */
	if (strlen(name) >= NAMEDATALEN)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
				 errmsg("invalid input value for enum %s: \"%s\"",
						format_type_be(enumtypoid),
						name)));

	tup = SearchSysCache2(ENUMTYPOIDNAME,
						  ObjectIdGetDatum(enumtypoid),
						  CStringGetDatum(name));
	if (!HeapTupleIsValid(tup))
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
				 errmsg("invalid input value for enum %s: \"%s\"",
						format_type_be(enumtypoid),
						name)));

	/* check it's safe to use in SQL */
	check_safe_enum_use(tup);

	enumoid = HeapTupleGetOid(tup);

	ReleaseSysCache(tup);

	pfree(name);

	PG_RETURN_OID(enumoid);
}

/*
 * to_regclass		- converts "classname" to class OID
 *
 * If the name is not found, we return NULL.
 */
Datum
to_regclass(PG_FUNCTION_ARGS)
{
	char	   *class_name = PG_GETARG_CSTRING(0);
	Oid			result;
	List	   *names;

	/*
	 * Parse the name into components and see if it matches any pg_class
	 * entries in the current search path.
	 */
	names = stringToQualifiedNameList(class_name);

	/* We might not even have permissions on this relation; don't lock it. */
	result = RangeVarGetRelid(makeRangeVarFromNameList(names), NoLock, true);

	if (OidIsValid(result))
		PG_RETURN_OID(result);
	else
		PG_RETURN_NULL();
}

Datum
enum_last(PG_FUNCTION_ARGS)
{
	Oid			enumtypoid;
	Oid			max = InvalidOid;
	CatCList   *list;
	int			num,
				i;

	/*
	 * We rely on being able to get the specific enum type from the calling
	 * expression tree.  Notice that the actual value of the argument isn't
	 * examined at all; in particular it might be NULL.
	 */
	enumtypoid = get_fn_expr_argtype(fcinfo->flinfo, 0);
	if (enumtypoid == InvalidOid)
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("could not determine actual enum type")));

	list = SearchSysCacheList(ENUMTYPOIDNAME, 1,
							  ObjectIdGetDatum(enumtypoid),
							  0, 0, 0);
	num = list->n_members;
	for (i = 0; i < num; i++)
	{
		Oid			valoid = HeapTupleHeaderGetOid(list->members[i]->tuple.t_data);

		if (!OidIsValid(max) || valoid > max)
			max = valoid;
	}

	ReleaseCatCacheList(list);

	if (!OidIsValid(max))		/* should not happen */
		elog(ERROR, "no values found for enum %s",
			 format_type_be(enumtypoid));

	PG_RETURN_OID(max);
}

Datum
enum_in(PG_FUNCTION_ARGS)
{
	char	   *name = PG_GETARG_CSTRING(0);
	Oid			enumtypoid = PG_GETARG_OID(1);
	Oid			enumoid;
	HeapTuple	tup;

	/* must check length to prevent Assert failure within SearchSysCache */
	if (strlen(name) >= NAMEDATALEN)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
				 errmsg("invalid input value for enum %s: \"%s\"",
						format_type_be(enumtypoid),
						name)));

	tup = SearchSysCache2(ENUMTYPOIDNAME,
						  ObjectIdGetDatum(enumtypoid),
						  CStringGetDatum(name));
	if (!HeapTupleIsValid(tup))
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
				 errmsg("invalid input value for enum %s: \"%s\"",
						format_type_be(enumtypoid),
						name)));

	/* check it's safe to use in SQL */
	check_safe_enum_use(tup);

	/*
	 * This comes from pg_enum.oid and stores system oids in user tables. This
	 * oid must be preserved by binary upgrades.
	 */
	enumoid = HeapTupleGetOid(tup);

	ReleaseSysCache(tup);

	PG_RETURN_OID(enumoid);
}

Datum
lo_create(PG_FUNCTION_ARGS)
{
	Oid			lobjId = PG_GETARG_OID(0);

#ifdef PGXC
	ereport(ERROR,
			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
			 errmsg("Postgres-XC does not support large object yet"),
			 errdetail("The feature is not currently supported")));
#endif

	/*
	 * We don't actually need to store into fscxt, but create it anyway to
	 * ensure that AtEOXact_LargeObject knows there is state to clean up
	 */
	CreateFSContext();

	lobjId = inv_create(lobjId);

	PG_RETURN_OID(lobjId);
}

/*
 * regprocin		- converts "proname" to proc OID
 *
 * We also accept a numeric OID, for symmetry with the output routine.
 *
 * '-' signifies unknown (OID 0).  In all other cases, the input must
 * match an existing pg_proc entry.
 */
Datum
regprocin(PG_FUNCTION_ARGS)
{
	char	   *pro_name_or_oid = PG_GETARG_CSTRING(0);
	RegProcedure result = InvalidOid;
	List	   *names;
	FuncCandidateList clist;

	/* '-' ? */
	if (strcmp(pro_name_or_oid, "-") == 0)
		PG_RETURN_OID(InvalidOid);

	/* Numeric OID? */
	if (pro_name_or_oid[0] >= '0' &&
		pro_name_or_oid[0] <= '9' &&
		strspn(pro_name_or_oid, "0123456789") == strlen(pro_name_or_oid))
	{
		result = DatumGetObjectId(DirectFunctionCall1(oidin,
										  CStringGetDatum(pro_name_or_oid)));
		PG_RETURN_OID(result);
	}

	/* Else it's a name, possibly schema-qualified */

	/*
	 * In bootstrap mode we assume the given name is not schema-qualified, and
	 * just search pg_proc for a unique match.	This is needed for
	 * initializing other system catalogs (pg_namespace may not exist yet, and
	 * certainly there are no schemas other than pg_catalog).
	 */
	if (IsBootstrapProcessingMode())
	{
		int			matches = 0;
		Relation	hdesc;
		ScanKeyData skey[1];
		SysScanDesc sysscan;
		HeapTuple	tuple;

		ScanKeyInit(&skey[0],
					Anum_pg_proc_proname,
					BTEqualStrategyNumber, F_NAMEEQ,
					CStringGetDatum(pro_name_or_oid));

		hdesc = heap_open(ProcedureRelationId, AccessShareLock);
		sysscan = systable_beginscan(hdesc, ProcedureNameArgsNspIndexId, true,
									 NULL, 1, skey);

		while (HeapTupleIsValid(tuple = systable_getnext(sysscan)))
		{
			result = (RegProcedure) HeapTupleGetOid(tuple);
			if (++matches > 1)
				break;
		}

		systable_endscan(sysscan);
		heap_close(hdesc, AccessShareLock);

		if (matches == 0)
			ereport(ERROR,
					(errcode(ERRCODE_UNDEFINED_FUNCTION),
				 errmsg("function \"%s\" does not exist", pro_name_or_oid)));

		else if (matches > 1)
			ereport(ERROR,
					(errcode(ERRCODE_AMBIGUOUS_FUNCTION),
					 errmsg("more than one function named \"%s\"",
							pro_name_or_oid)));

		PG_RETURN_OID(result);
	}

	/*
	 * Normal case: parse the name into components and see if it matches any
	 * pg_proc entries in the current search path.
	 */
	names = stringToQualifiedNameList(pro_name_or_oid);
	clist = FuncnameGetCandidates(names, -1, NIL, false, false, false);

	if (clist == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_FUNCTION),
				 errmsg("function \"%s\" does not exist", pro_name_or_oid)));
	else if (clist->next != NULL)
		ereport(ERROR,
				(errcode(ERRCODE_AMBIGUOUS_FUNCTION),
				 errmsg("more than one function named \"%s\"",
						pro_name_or_oid)));

	result = clist->oid;

	PG_RETURN_OID(result);
}

/*
 * Returns partition oid for specified parent relid and value.
 * In case when partition isn't exist try to create one.
 */
Datum
find_or_create_range_partition(PG_FUNCTION_ARGS)
{
	int		relid = DatumGetInt32(PG_GETARG_DATUM(0));
	Datum	value = PG_GETARG_DATUM(1);
	Oid		value_type = get_fn_expr_argtype(fcinfo->flinfo, 1);
	int		pos;
	bool	found;
	RangeRelation	*rangerel;
	RangeEntry		*ranges;
	TypeCacheEntry	*tce;
	PartRelationInfo *prel;
	Oid				 cmp_proc_oid;
	FmgrInfo		 cmp_func;

	tce = lookup_type_cache(value_type,
		TYPECACHE_EQ_OPR | TYPECACHE_LT_OPR | TYPECACHE_GT_OPR |
		TYPECACHE_CMP_PROC | TYPECACHE_CMP_PROC_FINFO);

	prel = get_pathman_relation_info(relid, NULL);
	rangerel = get_pathman_range_relation(relid, NULL);

	if (!prel || !rangerel)
		PG_RETURN_NULL();

	cmp_proc_oid = get_opfamily_proc(tce->btree_opf,
									 value_type,
									 prel->atttype,
									 BTORDER_PROC);
	fmgr_info(cmp_proc_oid, &cmp_func);

	ranges = dsm_array_get_pointer(&rangerel->ranges);
	pos = range_binary_search(rangerel, &cmp_func, value, &found);

	/*
	 * If found then just return oid. Else create new partitions
	 */
	if (found)
		PG_RETURN_OID(ranges[pos].child_oid);
	/*
	 * If not found and value is between first and last partitions
	*/
	if (!found && pos >= 0)
		PG_RETURN_NULL();
	else
	{
		Oid		child_oid;
		bool	crashed = false;

		/* Lock config before appending new partitions */
		LWLockAcquire(pmstate->load_config_lock, LW_EXCLUSIVE);

		/* Restrict concurrent partition creation */
		LWLockAcquire(pmstate->edit_partitions_lock, LW_EXCLUSIVE);

		/*
		 * Check if someone else has already created partition.
		 */
		ranges = dsm_array_get_pointer(&rangerel->ranges);
		pos = range_binary_search(rangerel, &cmp_func, value, &found);
		if (found)
		{
			LWLockRelease(pmstate->edit_partitions_lock);
			LWLockRelease(pmstate->load_config_lock);
			PG_RETURN_OID(ranges[pos].child_oid);
		}

		/* Start background worker to create new partitions */
		child_oid = create_partitions_bg_worker(relid, value, value_type, &crashed);

		// SPI_connect();
		// child_oid = create_partitions(relid, value, value_type, &crashed);
		// SPI_finish();
		// elog(WARNING, "Worker finished");

		/* Release locks */
		if (!crashed)
		{
			LWLockRelease(pmstate->edit_partitions_lock);
			LWLockRelease(pmstate->load_config_lock);
		}

		/* Repeat binary search */
		ranges = dsm_array_get_pointer(&rangerel->ranges);
		pos = range_binary_search(rangerel, &cmp_func, value, &found);
		if (found)
			PG_RETURN_OID(child_oid);
	}

	PG_RETURN_NULL();
}

/*
 * regtypein		- converts "typename" to type OID
 *
 * We also accept a numeric OID, for symmetry with the output routine.
 *
 * '-' signifies unknown (OID 0).  In all other cases, the input must
 * match an existing pg_type entry.
 *
 * In bootstrap mode the name must just equal some existing name in pg_type.
 * In normal mode the type name can be specified using the full type syntax
 * recognized by the parser; for example, DOUBLE PRECISION and INTEGER[] will
 * work and be translated to the correct type names.  (We ignore any typmod
 * info generated by the parser, however.)
 */
Datum
regtypein(PG_FUNCTION_ARGS)
{
	char	   *typ_name_or_oid = PG_GETARG_CSTRING(0);
	Oid			result = InvalidOid;
	int32		typmod;

	/* '-' ? */
	if (strcmp(typ_name_or_oid, "-") == 0)
		PG_RETURN_OID(InvalidOid);

	/* Numeric OID? */
	if (typ_name_or_oid[0] >= '0' &&
		typ_name_or_oid[0] <= '9' &&
		strspn(typ_name_or_oid, "0123456789") == strlen(typ_name_or_oid))
	{
		result = DatumGetObjectId(DirectFunctionCall1(oidin,
										  CStringGetDatum(typ_name_or_oid)));
		PG_RETURN_OID(result);
	}

	/* Else it's a type name, possibly schema-qualified or decorated */

	/*
	 * In bootstrap mode we assume the given name is not schema-qualified, and
	 * just search pg_type for a match.  This is needed for initializing other
	 * system catalogs (pg_namespace may not exist yet, and certainly there
	 * are no schemas other than pg_catalog).
	 */
	if (IsBootstrapProcessingMode())
	{
		int			matches = 0;

		result = 
				caql_getoid_plus(
						NULL,
						&matches,
						NULL,
						cql("SELECT oid FROM pg_type "
							" WHERE typname = :1 ",
							CStringGetDatum(typ_name_or_oid)));
		if (0 == matches)
		{
			ereport(ERROR,
					(errcode(ERRCODE_UNDEFINED_OBJECT),
					 errmsg("type \"%s\" does not exist", typ_name_or_oid)));
		}

		/* We assume there can be only one match */

		PG_RETURN_OID(result);
	}

	/*
	 * Normal case: invoke the full parser to deal with special cases such as
	 * array syntax.
	 */
	parseTypeString(typ_name_or_oid, &result, &typmod);

	PG_RETURN_OID(result);
}

/*
 * regclassin		- converts "classname" to class OID
 *
 * We also accept a numeric OID, for symmetry with the output routine.
 *
 * '-' signifies unknown (OID 0).  In all other cases, the input must
 * match an existing pg_class entry.
 */
Datum
regclassin(PG_FUNCTION_ARGS)
{
	char	   *class_name_or_oid = PG_GETARG_CSTRING(0);
	Oid			result = InvalidOid;
	List	   *names;

	/* '-' ? */
	if (strcmp(class_name_or_oid, "-") == 0)
		PG_RETURN_OID(InvalidOid);

	/* Numeric OID? */
	if (class_name_or_oid[0] >= '0' &&
		class_name_or_oid[0] <= '9' &&
		strspn(class_name_or_oid, "0123456789") == strlen(class_name_or_oid))
	{
		result = DatumGetObjectId(DirectFunctionCall1(oidin,
										CStringGetDatum(class_name_or_oid)));
		PG_RETURN_OID(result);
	}

	/* Else it's a name, possibly schema-qualified */

	/*
	 * In bootstrap mode we assume the given name is not schema-qualified, and
	 * just search pg_class for a match.  This is needed for initializing
	 * other system catalogs (pg_namespace may not exist yet, and certainly
	 * there are no schemas other than pg_catalog).
	 */
	if (IsBootstrapProcessingMode())
	{
		int			matches = 0;

		result = 
				caql_getoid_plus(
						NULL,
						&matches,
						NULL,
						cql("SELECT oid FROM pg_class "
							" WHERE relname = :1 ",
							CStringGetDatum(class_name_or_oid)));
		if (0 == matches)
		{
			ereport(ERROR,
					(errcode(ERRCODE_UNDEFINED_TABLE),
			   errmsg("relation \"%s\" does not exist", class_name_or_oid)));
		}

		/* We assume there can be only one match */

		PG_RETURN_OID(result);
	}

	/*
	 * Normal case: parse the name into components and see if it matches any
	 * pg_class entries in the current search path.
	 */
	names = stringToQualifiedNameList(class_name_or_oid, "regclassin");

	result = RangeVarGetRelid(makeRangeVarFromNameList(names), false, true /*allowHcatalog*/);

	PG_RETURN_OID(result);
}