This article provides a step-by-step guide for installing the RisingWave Java UDF SDK, defining functions using Java, starting a Java process as a UDF server, and declaring and using UDFs in RisingWave.
The RisingWave Java UDF SDK is distributed as a Maven artifact. We have prepared a sample project so you don’t have to create it from scratch. Run the following command to clone the template repository.
A user-defined scalar function maps zero, one, or multiple scalar values to a new scalar value.
In order to define a scalar function, you have to create a new class that implements the ScalarFunctioninterface in com.risingwave.functions and implement exactly one evaluation method named eval(...). This method must be declared public and non-static.
Any data type listed in Data type mapping can be used as a parameter or return type of an evaluation method.
Here’s an example of a scalar function that calculates the greatest common divisor (GCD) of two integers:
Copy
Ask AI
import com.risingwave.functions.ScalarFunction;public class Gcd implements ScalarFunction { public int eval(int a, int b) { while (b != 0) { int temp = b; b = a % b; a = temp; } return a; }}
DIFFERENCES WITH FLINK
The ScalarFunction is an interface instead of an abstract class.
Multiple overloaded eval methods are not supported.
Variable arguments such as eval(Integer...) are not supported.
A user-defined table function maps zero, one, or multiple scalar values to one or multiple rows (structured types).
In order to define a table function, you have to create a new class that implements the TableFunctioninterface in com.risingwave.functions and implement exactly one evaluation method named eval(...). This method must be declared public and non-static.
The return type must be an Iterator of any data type listed in Data type mapping.
Similar to scalar functions, input and output data types are automatically extracted using reflection. This includes the generic argument T of the return value for determining an output data type.
Here’s an example of a table function that generates a series of integers:
Copy
Ask AI
import com.risingwave.functions.TableFunction;public class Series implements TableFunction { public Iterator<Integer> eval(int n) { return java.util.stream.IntStream.range(0, n).iterator(); }}
DIFFERENCES WITH FLINK
The TableFunction is an interface instead of an abstract class. It has no generic arguments.
Instead of calling collect to emit a row, the eval method returns an Iterator of the output rows.
Multiple overloaded eval methods are not supported.
Variable arguments such as eval(Integer...) are not supported.
In SQL, table functions can be used in the FROM clause directly. JOIN LATERAL TABLE is not supported.
In RisingWave, use the CREATE FUNCTION command to declare the functions you defined.
Here are the SQL statements for declaring the two UDFs defined in step 3.
Copy
Ask AI
CREATE FUNCTION gcd(int, int) RETURNS intAS gcdUSING LINK 'http://localhost:8815';CREATE FUNCTION series(int) RETURNS TABLE (x int)AS seriesUSING LINK 'http://localhost:8815';
import com.google.gson.Gson;// Returns the i-th element of a JSON array.public class JsonbAccess implements ScalarFunction { static Gson gson = new Gson(); public @DataTypeHint("JSONB") String eval(@DataTypeHint("JSONB") String json, int index) { if (json == null) return null; var array = gson.fromJson(json, Object[].class); if (index >= array.length || index < 0) return null; var obj = array[index]; return gson.toJson(obj); }}
Create the function in RisingWave
Copy
Ask AI
CREATE FUNCTION jsonb_access(jsonb, int) RETURNS jsonbAS jsonb_access USING link 'http://localhost:8815';
// Split a socket address into host and port.public static class IpPort implements ScalarFunction { public static class SocketAddr { public String host; public short port; } public SocketAddr eval(String addr) { var socketAddr = new SocketAddr(); var parts = addr.split(":"); socketAddr.host = parts[0]; socketAddr.port = Short.parseShort(parts[1]); return socketAddr; }}
Create the function in RisingWave
Copy
Ask AI
CREATE FUNCTION ip_port(varchar) RETURNS struct<host varchar, port smallint>AS ip_port USING link 'http://localhost:8815';
Assistant
Responses are generated using AI and may contain mistakes.