[FLINK-27776] Throw exception when UDAF used in sliding window does not implement merge method in PyFlink

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.13.6, 1.14.4, 1.15.0
• Fix Version/s: 1.14.5, 1.15.1, 1.16.0
• Component/s: API / Python
• Labels: pull-request-available
• Assignee: Huang Xingbo
• Reporter: Huang Xingbo
• Created: 25/May/22 12:09
• Updated: 17/Jun/22 13:13
• Resolved: 26/May/22 08:38

Description

We use the pane state to optimize the result of calculating the window state, which requires udaf to implement the merge method. However, due to the lack of detection of whether the merge method of udaf is implemented, the user's output result did not meet his expectations and there is no exception. Below is an example of a UDAF that implements the merge method:

class SumAggregateFunction(AggregateFunction):

    def get_value(self, accumulator):
        return accumulator[0]

    def create_accumulator(self):
        return [0]

    def accumulate(self, accumulator, *args):
        accumulator[0] = accumulator[0] + args[0]

    def retract(self, accumulator, *args):
        accumulator[0] = accumulator[0] - args[0]

    def merge(self, accumulator, accumulators):
        for other_acc in accumulators:
            accumulator[0] = accumulator[0] + other_acc[0]

    def get_accumulator_type(self):
        return DataTypes.ARRAY(DataTypes.BIGINT())

    def get_result_type(self):
        return DataTypes.BIGINT()

Issue Links

• links to: GitHub Pull Request #19817