Java UDF

Develop a scalar UDF

A scalar UDF processes one row at a time, returning a single value for each row. Typical scalar functions include UPPER, LOWER, ROUND, and ABS.

The following example shows how to extract data from a JSON object. In some business scenarios, the value of a field in JSON data might be a JSON string instead of a JSON object. To extract the nested JSON string, you must use a nested call to the GET_JSON_STRING function, such as GET_JSON_STRING(GET_JSON_STRING('{"key":"{\\"k0\\":\\"v0\\"}"}', "$.key"), "$.k0").

To simplify the SQL statement, you can develop a UDF to extract the JSON string directly, for example, MY_UDF_JSON_GET('{"key":"{\\"k0\\":\\"v0\\"}"}', "$.key.k0").

package com.starrocks.udf.sample;
import com.alibaba.fastjson.JSONPath;

public class UDFJsonGet {
    public final String evaluate(String jsonObj, String key) {
        if (jsonObj == null || key == null) return null;
        try {
            // The JSONPath library can fully expand the path, even if a field value is a JSON-formatted string.
            return JSONPath.read(jsonObj, key).toString();
        } catch (Exception e) {
            return null;
        }
    }
}

Your custom class must implement the following method.

Develop a UDAF

A UDAF operates on a group of rows and returns a single value. Typical aggregate functions include SUM, COUNT, MAX, and MIN. These functions aggregate data from multiple rows within each GROUP BY group and output a single result for each group.

The following example uses the MY_SUM_INT function. Unlike the built-in SUM function, which returns a BIGINT value, MY_SUM_INT takes an INT argument and returns an INT.

package com.starrocks.udf.sample;

public class SumInt {
    public static class State {
        int counter = 0;
        public int serializeLength() { return 4; }
    }

    public State create() {
        return new State();
    }

    public void destroy(State state) {
    }

    public final void update(State state, Integer val) {
        if (val != null) {
            state.counter+= val;
        }
    }

    public void serialize(State state, java.nio.ByteBuffer buff) {
        buff.putInt(state.counter);
    }

    public void merge(State state, java.nio.ByteBuffer buffer) {
        int val = buffer.getInt();
        state.counter += val;
    }

    public Integer finalize(State state) {
        return state.counter;
    }
}

Your custom class must implement the following methods.

When developing a UDAF, you must use the java.nio.ByteBuffer buffer class to store the intermediate result and the serializeLength method to specify the length of the serialized intermediate result.

Develop a UDWF

A UDWF is a special type of aggregate function. Unlike a regular aggregate function, a window function calculates a value over a group of rows (a window) and returns a separate result for each row. Typically, a window function includes an OVER clause that divides rows into partitions. The function then computes a result for each row based on the window of rows it belongs to.

The following example uses the MY_WINDOW_SUM_INT function. Unlike the built-in SUM function, which returns a BIGINT value, MY_WINDOW_SUM_INT takes an INT argument and returns an INT.

package com.starrocks.udf.sample;

public class WindowSumInt {    
    public static class State {
        int counter = 0;
        public int serializeLength() { return 4; }
        @Override
        public String toString() {
            return "State{" +
                    "counter=" + counter +
                    '}';
        }
    }

    public State create() {
        return new State();
    }

    public void destroy(State state) {

    }

    public void update(State state, Integer val) {
        if (val != null) {
            state.counter+=val;
        }
    }

    public void serialize(State state, java.nio.ByteBuffer buff) {
        buff.putInt(state.counter);
    }

    public void merge(State state, java.nio.ByteBuffer buffer) {
        int val = buffer.getInt();
        state.counter += val;
    }

    public Integer finalize(State state) {
        return state.counter;
    }

    public void reset(State state) {
        state.counter = 0;
    }

    public void windowUpdate(State state,
                            int peer_group_start, int peer_group_end,
                            int frame_start, int frame_end,
                            Integer[] inputs) {
        for (int i = (int)frame_start; i < (int)frame_end; ++i) {
            state.counter += inputs[i];
        }
    }
}

Because a window function is a special type of aggregate function, your custom class must implement all the required UDAF methods, plus the windowUpdate() method.

Develop a UDTF

A UDTF takes a single row as input and produces a table of output rows. UDTFs are often used for operations like splitting a column into multiple rows.

The following example uses the MY_UDF_SPLIT function. It splits a string by using a space as the delimiter. Both the input argument and return value are of the STRING type.

package com.starrocks.udf.sample;

public class UDFSplit{
    public String[] process(String in) {
        if (in == null) return null;
        return in.split(" ");
    }
}

Your custom class must implement the following method.

Syntax

CREATE [GLOBAL][AGGREGATE | TABLE] FUNCTION function_name(arg_type [, ...])
RETURNS return_type
[PROPERTIES ("key" = "value" [, ...]) ]

Parameters

Create a scalar UDF

Run the following command to create the scalar UDF from the previous example in StarRocks.

CREATE [GLOBAL] FUNCTION MY_UDF_JSON_GET(string, string) 
RETURNS string
PROPERTIES (
    "symbol" = "com.starrocks.udf.sample.UDFJsonGet", 
    "type" = "StarrocksJar",
    "file" = "http://<YourBucketName>.oss-cn-xxxx-internal.aliyuncs.com/<YourPath>/udf-1.0-SNAPSHOT-jar-with-dependencies.jar"
);

Create a UDAF

Run the following command to create the UDAF from the previous example in StarRocks.

CREATE [GLOBAL] AGGREGATE FUNCTION MY_SUM_INT(INT) 
RETURNS INT
PROPERTIES 
( 
    "symbol" = "com.starrocks.udf.sample.SumInt", 
    "type" = "StarrocksJar",
    "file" = "http://<YourBucketName>.oss-cn-xxxx-internal.aliyuncs.com/<YourPath>/udf-1.0-SNAPSHOT-jar-with-dependencies.jar"
);

The parameters in the PROPERTIES clause are the same as those described in Create a scalar UDF.

Create a UDWF

Run the following command to create the UDWF from the previous example in StarRocks.

CREATE [GLOBAL] AGGREGATE FUNCTION MY_WINDOW_SUM_INT(Int)
RETURNS Int
PROPERTIES 
(
    "analytic" = "true",
    "symbol" = "com.starrocks.udf.sample.WindowSumInt", 
    "type" = "StarrocksJar", 
    "file" = "http://<YourBucketName>.oss-cn-xxxx-internal.aliyuncs.com/<YourPath>/udf-1.0-SNAPSHOT-jar-with-dependencies.jar"
);

The analytic property identifies the function as a window function. For a UDWF, set this to true. The other parameters are the same as those described in Create a scalar UDF.

Create a UDTF

Run the following command to create the UDTF from the previous example in StarRocks.

CREATE [GLOBAL] TABLE FUNCTION MY_UDF_SPLIT(string)
RETURNS string
PROPERTIES 
(
    "symbol" = "com.starrocks.udf.sample.UDFSplit", 
    "type" = "StarrocksJar", 
    "file" = "http://<YourBucketName>.oss-cn-xxxx-internal.aliyuncs.com/<YourPath>/udf-1.0-SNAPSHOT-jar-with-dependencies.jar"
);

The parameters in the PROPERTIES clause are the same as those described in Create a scalar UDF.

Use the scalar UDF

Run the following command to use the scalar UDF created in Step 6.

SELECT MY_UDF_JSON_GET('{"key":"{\\"in\\":2}"}', '$.key.in');

Use the UDAF

Run the following command to use the UDAF created in Step 6.

SELECT MY_SUM_INT(col1);

Use the UDWF

Run the following command to use the UDWF created in Step 6.

SELECT MY_WINDOW_SUM_INT(intcol) 
            OVER (PARTITION BY intcol2
                  ORDER BY intcol3
                  ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)
FROM test_basic;

Use the UDTF

Run the following command to use the UDTF from the previous example.

-- Assume that table t1 exists with columns a, b, and c1.
SELECT t1.a,t1.b,t1.c1 FROM t1;
> output:
1,2.1,"hello world"
2,2.2,"hello UDTF."

-- Use the MY_UDF_SPLIT() function.
SELECT t1.a,t1.b, MY_UDF_SPLIT FROM t1, MY_UDF_SPLIT(t1.c1); 
> output:
1,2.1,"hello"
1,2.1,"world"
2,2.2,"hello"
2,2.2,"UDTF."