Base class for a user-defined asynchronously table function (UDTF). This is similar to
TableFunction
but this function is asynchronously.
A user-defined table functions works on
zero, one, or multiple scalar values as input and returns multiple rows as output.
The behavior of a
AsyncTableFunction
can be defined by implementing a custom evaluation
method. An evaluation method must be declared publicly, not static and named "eval".
Evaluation methods can also be overloaded by implementing multiple methods named "eval".
The first parameter of evaluation method must be
CompletableFuture
, and the others are
user defined input parameters like the "eval" method of
TableFunction
. The generic type of
CompletableFuture
must be
Collection
to collect multiple possible result
values.
For each "eval", an async io operation can be triggered, and once it has been done,
the result can be collected by calling
CompletableFuture.complete(T)
. For each async
operation, its context is stored in the operator immediately after invoking "eval",
avoiding blocking for each stream input as long as the internal buffer is not full.
CompletableFuture
can be passed into callbacks or futures to collect the result data.
An error can also be propagate to the async IO operator by
CompletableFuture.completeExceptionally(Throwable)
.
User-defined functions must have a default constructor and must be instantiable during
runtime.
By default the result type of an evaluation method is determined by Flink's type extraction
facilities. Currently, only support
Row
and
RowData
as
the result type. Will support more complex, custom types in the future.
Example:
public class HBaseAsyncTableFunction extends AsyncTableFunction<String> {
// implement an "eval" method with as many parameters as you want
public void eval(CompletableFuture<Collection<String>> result, String rowkey) {
Get get = new Get(Bytes.toBytes(rowkey));
ListenableFuture<Result> future = hbase.asyncGet(get);
Futures.addCallback(future, new FutureCallback<Result>() {
public void onSuccess(Result result) {
List<String> ret = process(result);
result.complete(ret);
public void onFailure(Throwable thrown) {
result.completeExceptionally(thrown);
// you can overload the eval method here ...
NOTE: the
AsyncTableFunction
can not be used as UDTF currently. It only used in
temporal table join as an async lookup function.
See Also:
Serialized Form
getResultType
public TypeInformation<T> getResultType()
Returns the result type of the evaluation method with a given signature.
This method needs to be overridden in case Flink's type extraction facilities are not
sufficient to extract the
TypeInformation
based on the return type of the evaluation
method. Flink's type extraction facilities can handle basic types or
simple POJOs but might be wrong for more complex, custom, or composite types.
TypeInformation
of result type or
null
if Flink should determine the type
getKind
public final FunctionKind getKind()
Returns the kind of function this definition describes.
getTypeInference
public TypeInference getTypeInference(DataTypeFactory typeFactory)
Returns the logic for performing type inference of a call to this function definition.
The type inference process is responsible for inferring unknown types of input arguments,
validating input arguments, and producing result types. The type inference process happens
independent of a function body. The output of the type inference is used to search for a
corresponding runtime implementation.
Instances of type inference can be created by using
TypeInference.newBuilder()
.
See
BuiltInFunctionDefinitions
for concrete usage examples.
The type inference for user-defined functions is automatically extracted using reflection.
It does this by analyzing implementation methods such as
eval() or accumulate()
and
the generic parameters of a function class if present. If the reflective information is not
sufficient, it can be supported and enriched with
DataTypeHint
and
FunctionHint
annotations.
Note: Overriding this method is only recommended for advanced users. If a custom type
inference is specified, it is the responsibility of the implementer to make sure that the
output of the type inference process matches with the implementation method:
The implementation method must comply with each
DataType.getConversionClass()
returned by the type inference. For example, if
DataTypes.TIMESTAMP(3).bridgedTo(java.sql.Timestamp.class)
is an expected argument type, the
method must accept a call
eval(java.sql.Timestamp)
.
Regular Java calling semantics (including type widening and autoboxing) are applied when
calling an implementation method which means that the signature can be
eval(java.lang.Object)
.
The runtime will take care of converting the data to the data format specified by the
DataType.getConversionClass()
coming from the type inference logic.
getTypeInference
in interface
FunctionDefinition
Specified by:
getTypeInference
in class
UserDefinedFunction
