[jira] [Created] (FLINK-19592) MiniBatchGroupAggFunction should emit messages to prevent too early state eviction of downstream operators

["Smile (Jira)" <ji...@apache.org>]

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)