[FLINK-34876][transform] Support UDF functions in transform

[yuxiqian]

This closes FLINK-34876, aims to introduce customized logic into transform expressions, with some key features:

• supports UDF classes implemented in Java and Scala
• explicit CDC pipeline type hints
• open & close lifecycle hooks
• compatible with simple Flink ScalarFunction

---

One may add a UDF block into existing pipeline definition and call them in transform expressions:

transform:
  - projection: addone(addone(addone(addone(1))))
    # ...

pipeline:
  user-defined-function:
    - name: addone                                   # function name used in expressions
      classpath: org.example.udf.AddOneFunctionClass # fully-qualified class name
    - name: format
      classpath: org.example.udf.FormatFunctionClass

where classpath value either implements CDC's org.apache.flink.cdc.common.udf.UserDefinedFunction interface or extends Flink org.apache.flink.table.functions.ScalarFunction abstract class.

UserDefinedFunction is declared as follows:

public interface UserDefinedFunction {

    default DataType getReturnType() { return null; }

    default void open() throws Exception {}

    default void close() throws Exception {}
}

A valid UserDefinedFunction implementation must:

• Have a nullary constructor
• Have (at least one) eval method that doesn't return void

A UserDefinedFunction implementation may:

• Override getReturnType method to declare its return type.
  • It will be inferred from eval signature if not provided.

public class TypeHintFunctionClass implements UserDefinedFunction {

    @Override
    public DataType getReturnType() {
        return DataTypes.STRING();
    }

    public Object eval() {
        // Return type could not be inferred from function signature
        return "Forty-two";
    }
}

• Override open and close lifecycle hooks to initialize / clean-up necessary resources.
  • props defined in YAML will be passed in UserDefinedFunctionContext as an argument.
  • UDF's lifecycle is identical to TransformDataOperator (or PostTransformOperator after FLINK-35272).
  • UserDefinedFunction might be constructed temporarily elsewhere, so make sure resources are handled in open and close instead of constructors.

public class LifecycleFunctionClass implements UserDefinedFunction {

    private Integer counter;

    public String eval() {
        return "#" + (counter++); // This returns #1, #2, #3, ...
    }

    @Override
    public void open() {
        counter = 0;
        System.out.println("[ LifeFun ] opened.");
    }

    @Override
    public void close() {
        System.out.println("[ LifeFun ] closed. Called " + counter + " times.");
    }
}

eval method overloading is supported, and runtime invocations will be dispatched to the correct one:

public class TypeOfFunctionClass implements UserDefinedFunction {

    public String eval(Boolean b) {
        return "Boolean: " + b;
    }

    public String eval(Integer i) {
        return "Integer: " + i;
    }

    public String eval(Float f) {
        return "Float: " + f;
    }

    public String eval(Double d) {
        return "Double: " + d;
    }

    public String eval(String s) {
        return "String: " + s;
    }
}

Flink ScalarFunction is supported with the following limitations:

• Flink-style type hint (TypeInformation) will be ignored.
• ScalarFunction with arguments are not supported.
• open and close methods declared will be ignored.

[yuxiqian]

Having some difficulties getting Calcite related materials, maybe @aiwenmo could provide some expert advice?

[yuxiqian]

Adjusted based on previous discussions, cleaned up changes, and added UDF-related docs.

[yuxiqian]

Rebased to latest master, cc @ruanhang1993

Minor concern: By adding more E2e test cases, now it takes nearly 1h 30min to finish running e2e case, which is worrying since current timeout limit is configured to 90 minutes. Maybe something like #3514 is necessary?

[ruanhang1993]

Suggested change

	default DataType getReturnType() {
	return null;
	}
	default Optional<DataType> getReturnType() {
	return Optional.empty();
	}

[yuxiqian]

I have some doubt on this. This is meant to be a interface protocol, and any reasonable overwrite of this method should not return an empty value.

[yuxiqian]

Seems CI is failing due to an expired link in Doris docs. Rebased & fixed.

[yuxiqian]

Rebased with master and resolved conflicts. cc @leonardBang @ruanhang1993

[aiwenmo]

LGTM

[mamineturki]

@yuxiqian Hi, I have been trying to run an example with this new feature but I always get java.lang.ClassNotFoundException: org.example.udf.AddOneFunctionClass

I tried loading the udf function using the path in the PR description, as well as the package name org.apache.flink.cdc.udf.examples.scala.AddOneFunctionClass with no luck.

I imported the function by downloading the pre-packaged jar wget https://repo1.maven.org/maven2/org/apache/flink/flink-cdc-pipeline-udf-examples/3.2.0/flink-cdc-pipeline-udf-examples-3.2.0.jar

please advise on what am I missing exactly. Thanks

[yuxiqian]

Hi @mamineturki, flink-cdc-pipeline-udf-examples are for testing purposes only. If you'd like to test it, using org.apache.flink.cdc.udf.examples.scala.AddOneFunctionClass should work. Seems it wasn't successfully loaded by Flink classloaders in your case.

Have you tried putting .jar file into Flink's /lib path (not Flink CDC /lib!) and restart the Flink cluster? If that doesn't work, try passing it with --jar XXX.jar argument when submitting pipeline job with bin/flink-cdc.sh.

[mamineturki]

Thanks for the speedy response @yuxiqian . passing the jar using --jar worked for me.

[mamineturki]

Hi @yuxiqian I have another question please.
I was able to produce custom UDF functions that take one, two or multiple arguments. what I am looking to test next is a UDF that generates multiple columns as output (for example splitting a full name into first and last or one hot encoding of a column)
is this currently supported? Thanks,

[yuxiqian]

Hi @mamineturki, unfortunately it's not possible now, as YAML UDF is basically equivalent to Flink's ScalarFunction.

(For TableFunction or AggregationFunction support, feel free to create another feature request ticket on JIRA.)