xuyangzhong commented on code in PR #24051:
URL: https://github.com/apache/flink/pull/24051#discussion_r1448431727
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/ImperativeAggCodeGen.scala:
##########
@@ -83,7 +87,8 @@ class ImperativeAggCodeGen(
hasNamespace: Boolean,
mergedAccOnHeap: Boolean,
mergedAccExternalType: DataType,
- inputFieldCopy: Boolean)
+ inputFieldCopy: Boolean,
+ val isIncrementalUpdateNeeded: Boolean)
Review Comment:
nit: remove `val` here because the default attribute in construct is `val`.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/ImperativeAggCodeGen.scala:
##########
@@ -488,10 +493,14 @@ class ImperativeAggCodeGen(
}
if (needEmitValue) {
+ val (emitMethod, collectorClass) =
+ if (isIncrementalUpdateNeeded)
+ (UserDefinedFunctionHelper.TABLE_AGGREGATE_EMIT_RETRACT,
classOf[RetractableCollector[_]])
Review Comment:
nit: add braces for `If` block to resolve the warning
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/AggsHandlerCodeGenerator.scala:
##########
@@ -447,6 +473,23 @@ class AggsHandlerCodeGenerator(
val recordInputName = newName("recordInput")
val recordToRowDataCode = genRecordToRowData(aggExternalType,
recordInputName)
+ // for emitUpdateWithRetract, the collector needs to implement
RetractableCollector
+ // and override retract method
+ val (collectorClassName, collectorRetractCode) =
+ if (isIncrementalUpdateNeeded)
Review Comment:
ditto
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedTableFunctions.java:
##########
@@ -166,4 +169,91 @@ public void eval(String str) {
str.length() + 5));
}
}
+
+ /** Mutable accumulator of structured type for the table aggregate
function. */
Review Comment:
nit: add class JavaUserDefinedTableAggFunctions.java to restore these
function (+1), or just using UserDefinedTableAggFunctions.scala.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
