[FLINK-7101] Fix Non-windowed group-aggregate error when using `minIdleStateRetentionTime` config and retract agg

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.3.0, 1.3.1
• Fix Version/s: 1.4.0
• Component/s: Table SQL / API
• Assignee: sunjincheng
• Reporter: sunjincheng
• Created: 04/Jul/17 12:36
• Updated: 17/Jul/17 14:34
• Resolved: 17/Jul/17 14:34

Description

When Non-windowed group-aggregate using minIdleStateRetentionTime config and retract AGG, Will emit "NULL" agg value which we do not expect.
For example: (IntSumWithRetractAggFunction)
1. Receive: CRow(Row.of(6L: JLong, 5: JInt, "aaa"), true)
2. Cleanup state
3. Receive: CRow(Row.of(6L: JLong, 5: JInt, "aaa"), false) // acc.f1 = -1, getValue= null

So, we must change the logic of GroupAggProcessFunction as follows:

if (inputCnt != 0) {
     ...
    } else {
     ...
    }

TO

if (inputCnt > 0) {
     ...
    } else {
    if( null != prevRow.row){
     ...
     }
    }

In this case, the result will bigger than expected, but i think it's make sense, because user want cleanup state.(they should know the impact)
What do you think? fhueske hequn8128

Issue Links

• links to: GitHub Pull Request #4348