Smile created FLINK-19592:
-----------------------------
Summary: MiniBatchGroupAggFunction should emit messages to prevent
too early state eviction of downstream operators
Key: FLINK-19592
URL: https://issues.apache.org/jira/browse/FLINK-19592
Project: Flink
Issue Type: Bug
Components: Table SQL / Runtime
Affects Versions: 1.12.0
Reporter: Smile
Currently,
[GroupAggFunction|https://github.com/apache/flink/blob/master/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/aggregate/GroupAggFunction.java#L183]
will emit a retract and a new insert message when a new message with the same
key arrives. According to
[Flink-8566|https://issues.apache.org/jira/browse/FLINK-8566], it's a feature
to prevent too early state eviction of downstream operators.
However,
[MiniBatchGroupAggFunction|https://github.com/apache/flink/blob/master/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/aggregate/MiniBatchGroupAggFunction.java#L206]
doesn't. Before [Flink-8566|https://issues.apache.org/jira/browse/FLINK-8566]
being resolved, it should also emit these messages.
*GroupAggFunction.java:*
{code:java}
if (!stateCleaningEnabled && equaliser.equals(prevAggValue, newAggValue)) {
// newRow is the same as before and state cleaning is not enabled.
// We do not emit retraction and acc message.
// If state cleaning is enabled, we have to emit messages to prevent
too early
// state eviction of downstream operators.
return;
} else {
// retract previous result
if (generateUpdateBefore) {
// prepare UPDATE_BEFORE message for previous row
resultRow.replace(currentKey,
prevAggValue).setRowKind(RowKind.UPDATE_BEFORE);
out.collect(resultRow);
}
// prepare UPDATE_AFTER message for new row
resultRow.replace(currentKey,
newAggValue).setRowKind(RowKind.UPDATE_AFTER);
}
{code}
*MiniBatchGroupAggFunction.java:*
{code:java}
if (!equaliser.equals(prevAggValue, newAggValue)) {
// new row is not same with prev row
if (generateUpdateBefore) {
// prepare UPDATE_BEFORE message for previous row
resultRow.replace(currentKey,
prevAggValue).setRowKind(RowKind.UPDATE_BEFORE);
out.collect(resultRow);
}
// prepare UPDATE_AFTER message for new row
resultRow.replace(currentKey,
newAggValue).setRowKind(RowKind.UPDATE_AFTER);
out.collect(resultRow);
}
// new row is same with prev row, no need to output
{code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
