Create FlinkSQL UDF with generic return type

user_4396113 0
Viewed 488

I would like to define function MAX_BY that takes value of type T and ordering parameter of type Number and returns max element from window according to ordering (of type T). I've tried

public class MaxBy<T> extends AggregateFunction<T, Tuple2<T, Number>> {

    @Override
    public T getValue(Tuple2<T, Number> tuple) {
        return tuple.f0;
    }

    @Override
    public Tuple2<T, Number> createAccumulator() {
        return Tuple2.of(null, 0L);
    }

    public void accumulate(Tuple2<T, Number> acc, T value, Number order) {
        if (order.doubleValue() > acc.f1.doubleValue()) {
            acc.f0 = value;
            acc.f1 = order;
        }
    }
}

but I cannot register such function using TableEnvironment.registerFunction. Underneath Flink uses TypeInformation to match types within SQL query and with such definition it cannot determine types (at least that's what I suppose). I saw that it is possible to provide several accumulate functions but still - I think return type must be same for each overloaded method.

Built-in aggregation functions work similarly to what I want to achieve - MAX can take arbitrary column type and return the same type. That's why I suppose I should be able to do it as well.

flink-sql apache-flink
1 Answers

Unfortunately, Flink doesn't support aggregation functions with flexible return types. For the MAX function, the internal implementation defines the core logic independent of the the type and then creates an implementation for every supported type (see code).

Internally, MAX is then mapped to the right implementation, depending on the type.

I don't think that's possible if you define and register a function as user-defined aggregation function.

user_36095710
Related