[FEATURE] Calcite Engine Framework: General function (UDF/UDAF) framework

[LantaoJin]

Is your feature request related to a problem?
Framework work of the development feature of Calcite Engine.

What solution would you like?
General function (UDF/UDAF) framework: Built-in UDF/UDAF, Custom UDF/UDAF

[andy-k-improving]

Question mainly on the registration part, currently we have issue similar to #811 which prevent us to implement Storage engine specific function (Ex: OpenSearch) in concise manner and without workaround.

This is not a must for Calcite UDF integration, but definitely would be great and we should take this consideration when working on this item.

[LantaoJin]

Question mainly on the registration part, currently we have issue similar to #811 which prevent us to implement Storage engine specific function (Ex: OpenSearch) in concise manner and without workaround.

This is not a must for Calcite UDF integration, but definitely would be great and we should take this consideration when working on this item.

Thanks for the extra context. Yes, register UDF dynamically should be considered in design. BTW, in first stage about Calcite integration, we will implement it in OpenSearch storage engine only. But we should keep the interface easy to extend when we move to next stage (supporting more storage engines).

[xinyual]

Based on our discussions, we are currently focusing on two types of functions:

1. Calcite built-in functions
2. PPL built-in functions that are not included in Calcite's built-in functions

Our architecture follows these steps:

1. Parse PPL into a Concrete Syntax Tree (CST).
2. Use a node visitor to:
3. Map non-aggregate functions to RexNode.
4. Map aggregate functions to AggCall.
   For Calcite built-in functions (Part 1), we can leverage RelBuilder's built-in functions and SqlStdOperatorTable's built-in operations. The primary task here is to map PPL functions to their corresponding Calcite functions.

For PPL-specific functions not in Calcite (Part 2), we need to implement our own built-in and aggregation functions. The simplest approach to align with the existing logic is to wrap our custom functions into the appropriate Calcite class.

UDF



For user defined non-aggregate function, we define the interface

public interface UserDefinedFunction  {
    Object eval(Object... args);
}

All userDefinedFunction should implement the function eval , which will take the list of object as input arguments, containing field value and arguments. For example, if we implement ltrim function, the eval function should be

public String eval(Object... args) {
        List<Object> argList = List.of(args);
        String field = argList.get(0).toString();
        int number = Integer.parseInt(argList.get(1).toString());
        return field.substring(0, number);
    }

Also, We provide a function in UserDefineFunctionUtils to wrap this class to a RexNode by

public static SqlOperator TransferUserDefinedFunction(
            Class<?> UDF,
            String functionName,
            SqlReturnTypeInference returnType
    ){
        final ScalarFunction udfFunction = ScalarFunctionImpl.create(Types.lookupMethod(UDF, "eval", Object[].class));
        SqlIdentifier udfLtrimIdentifier = new SqlIdentifier(Collections.singletonList(functionName), null, SqlParserPos.ZERO, null);
        return new SqlUserDefinedFunction(
                udfLtrimIdentifier,
                SqlKind.OTHER_FUNCTION,
                returnType,
                null,
                null,
                udfFunction);
    }

UDAF

For user defined aggregation function, we create two interface
Accumulator

public interface Accumulator {
    Object result();
}

And userDefinedAggFunction

public interface UserDefinedAggFunction<S extends Accumulator> {
    // return the accumulator
    S init();
    
    // Get the result of accumulator
    Object result(S accumulator);

    // Add values to the accumulator
    S add(S acc, Object... values);
}

The accumulator will accept the field value and then call result to output the result. And each UserDefinedAggFunction should have its own accumulator, when init it will return the accumulator, then for each field in aggregation process, it will call add and then finally it will call the result to get the final answer.
We provide a function in UserDefineFunctionUtils to wrap this class to an AggCall by

public static RelBuilder.AggCall TransferUserDefinedAggFunction(
            Class<? extends UserDefinedAggFunction> UDAF,
            String functionName,
            SqlReturnTypeInference returnType,
            List<RexNode> fields,
            List<RexNode> argList,
            RelBuilder relBuilder
    ){
        SqlUserDefinedAggFunction sqlUDAF = new SqlUserDefinedAggFunction(
                new SqlIdentifier(functionName, SqlParserPos.ZERO),
                SqlKind.OTHER_FUNCTION,
                returnType,
                null,
                null,
                AggregateFunctionImpl.create(UDAF),
                false,
                false,
                Optionality.FORBIDDEN
        );
        List<RexNode> addArgList = new ArrayList<>(fields);
        addArgList.addAll(argList);
        return relBuilder.aggregateCall(sqlUDAF, addArgList);
    }



[xinyual]

I compare the function difference for current index PPL and spark PPL. And we will try to first match them to calcite built-in functions, if not, we will implement our logic using the above logic

type	function name	index	spark	Map to calcite
CMD	stats	Y	Y
	patterns	Y	Y
	top	Y	Y
	parse	Y	Y
	sort	Y	Y
	dedup	Y	Y
	rare	Y	Y
	describe	Y	Y
	fields	Y	Y
	where	Y	Y
	eval	Y	Y
	head	Y	Y
	trendline	N	Y
	tpch	N	Y
	subquery	N	Y
	search	N	Y
	lookup	N	Y
	join	N	Y
	fillnull	N	Y
	flatten	N	Y
	fieldsummary	N	Y
	expand	N	Y
AGG	count	Y	Y
	sum	Y	Y
	avg	Y	Y
	max	Y	Y
	min	Y	Y
	var_samp	Y	Y
	var_pop	Y	Y
	stddev_samp	Y	N
	stddev_poo	Y	N
	take	Y	Y
	percentile	Y	Y
Datetime	addtime	Y	Y
	now	Y	Y
	date_sub	Y	Y
	date_add	Y	Y
	date	Y	Y
	hour	Y	Y
	month	Y	Y
	curdate	Y	Y
	date_format	Y	Y
	datediff	Y	Y
	day_of_week	Y	Y
	day_of_year	Y	Y
	dayname	Y	Y
	EARLIEST	N	Y
	LATEST	N	Y
	RELATIVE_TIMESTAMP	N	Y
	extract	Y	Y
	from_unixtime	Y	Y
	last_day	Y	Y
	makedate	Y	Y
	minute	Y	Y
	monthname	Y	Y
	quarter	Y	Y
	timestamp	Y	Y
	second	Y	Y
	convert_tz	Y	N
	datetime	Y	N
	datetime	Y	N
	from_days	Y	N
	get_format	Y	N
	maketime	Y	N
	microsecond	Y	N
	period_diff	Y	N
	period_add	Y	N
	str_to_date	Y	N
	subtime	Y	N
	sysdate	Y	N
	time	Y	N
	time_format	Y	N
	time_to_sec	Y	N
	timediff	Y	N
	sec_to_time	Y	N
	timestampadd	Y	N
	timestampdiff	Y	N
	to_days	Y	N
	to_seconds	Y	N
	unix_timestamp	Y	Y
	utc_date	Y	Y
	utc_time	Y	Y
	utc_timestamp	Y	Y
	week	Y	Y
	weekday	Y	Y
	year	Y	Y
	yearweek	Y	Y
Math	abs	Y	Y	SqlStdOperatorTable.ABS
	acos	Y	Y	SqlStdOperatorTable.ACOS
	asin	Y	Y	SqlStdOperatorTable.ASIN
	atan	Y	Y	SqlStdOperatorTable.ATAN2 with argument modification
	atan2	Y	Y	SqlStdOperatorTable.ATAN2
	cbrt	Y	Y	SqlStdOperatorTable.CBRT
	ceiling	Y	Y	SqlStdOperatorTable.CEIL
	conv	Y	Y	SqlStdOperatorTable.CONVERT ???
	cos	Y	Y	SqlStdOperatorTable.COS
	cot	Y	Y	SqlStdOperatorTable.COT
	crc32	Y	Y	no implementation
	degrees	Y	Y	SqlStdOperatorTable.DEGREES
	e	Y	Y	no implementation
	exp	Y	Y	SqlStdOperatorTable.EXP
	floor	Y	Y	SqlStdOperatorTable.FLOOR
	ln	Y	Y	SqlStdOperatorTable.LN
	log	Y	Y	SqlStdOperatorTable.LOG and reverse argument
	log2			SqlLibraryOperators.LOG2
	log10			SqlStdOperatorTable.LOG10
	mod	Y	Y	no implementation since we support double double
	pi	Y	Y	SqlStdOperatorTable.PI
	pow	Y	Y	SqlStdOperatorTable.POWER
	radians	Y	Y	SqlStdOperatorTable.RADIANS
	rand	Y	Y	SqlStdOperatorTable.RAND
	round	Y	Y	SqlStdOperatorTable.ROUND
	sign	Y	Y	SqlStdOperatorTable.SIGN
	sin	Y	Y	SqlStdOperatorTable.SIN
	sqrt	Y	Y	no implemtation
string	concat	Y	Y	SqlLibraryOperators.CONCAT_FUNCTION
	like	Y	Y	SqlLibraryOperators.ILIKE
	concat_ws	Y	Y	SqlLibraryOperators.CONCAT_WS
	length	Y	Y	SqlLibraryOperators.LENGTH
	lower	Y	Y	SqlStdOperatorTable.LOWER
	ltrim	Y	Y	SqlStdOperatorTable.TRM but change argument
	position	Y	Y	SqlStdOperatorTable.POSITION
	reverse	Y	Y	SqlLibraryOperators.REVERSE
	right	Y	Y	SqlLibraryOperators.RIGHT
	rtrim	Y	Y	SqlStdOperatorTable.TRM but change argument
	substring	Y	Y	SqlStdOperatorTable.SUBSTRING
	trim	Y	Y	SqlStdOperatorTable.TRM but change argument
	upper	Y	Y	SqlStdOperatorTable.UPPER
condition	ifnull	Y	Y	no implementation
	if	Y	Y	no implementation
	isnotnull	Y	Y
	isnull	Y	Y
	nullif	Y	Y	no implementation
relevance	match	Y	N
	query_string	Y	N
	match_bool_prefix	Y	N
	match_phrase	Y	N
	match_phrase_prefix	Y	N
	multi_match	Y	N
	simple_query_string	Y	N
System	typeof	Y	Y
conversion	cast	Y	Y
IP	CIDRMATCH	N	Y
	GEOIP	N	Y
Json	json	N	Y
	json_object	N	Y
	json_array	N	Y
	to_json_string	N	Y
	JSON_ARRAY_LENGTH	N	Y
	JSON_EXTRACT	N	Y
	JSON_DELETE	N	Y
	JSON_APPEND	N	Y
	JSON_KEYS	N	Y
	JSON_VALID	N	Y
	FORALL	N	Y
	EXISTS	N	Y
	FILTER	N	Y
	TRANSFORM	N	Y
	REDUCE	N	Y
Collection	array	N	Y
	ARRAY_LENGTH	N	Y
	FORALL	N	Y
	EXISTS	N	Y
	FILTER	N	Y
	TRANSFORM	N	Y
	REDUCE	N	Y
Cryptographic	MD5	N	Y
	SHA1	N	Y
	SHA2	N	Y