JingsongLi commented on a change in pull request #9322: [FLINK-13471][table]
Add FlatAggregate support to stream Table API(blink planner)
URL: https://github.com/apache/flink/pull/9322#discussion_r310109577
##########
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/AggsHandlerCodeGenerator.scala
##########
@@ -397,6 +398,146 @@ class AggsHandlerCodeGenerator(
new GeneratedAggsHandleFunction(functionName, functionCode,
ctx.references.toArray)
}
+ /**
+ * Generate [[GeneratedTableAggsHandleFunction]] with the given function
name and aggregate
+ * infos.
+ */
+ def generateTableAggsHandler(
+ name: String,
+ aggInfoList: AggregateInfoList): GeneratedTableAggsHandleFunction = {
+
+ initialAggregateInformation(aggInfoList)
+
+ // generates all methods body first to add necessary reuse code to context
+ val createAccumulatorsCode = genCreateAccumulators()
+ val getAccumulatorsCode = genGetAccumulators()
+ val setAccumulatorsCode = genSetAccumulators()
+ val resetAccumulatorsCode = genResetAccumulators()
+ val accumulateCode = genAccumulate()
+ val retractCode = genRetract()
+ val mergeCode = genMerge()
+ val emitValueCode = genEmitValue()
+
+ // gen converter
+ val aggExternalType =
aggInfoList.getActualAggregateInfos(0).externalResultType
+ val collectorInputName = newName("collectorInput")
+ val converterCode = CodeGenUtils.genToInternal(ctx, aggExternalType,
collectorInputName)
+
+ val functionName = newName(name)
+ val functionCode =
+ j"""
+ public final class $functionName implements
$TABLE_AGGS_HANDLER_FUNCTION {
+
+ ${ctx.reuseMemberCode()}
+ private $CONVERT_COLLECTOR_TYPE_TERM $MEMBER_COLLECTOR_TERM;
+
+ public $functionName(java.lang.Object[] references) throws Exception
{
+ ${ctx.reuseInitCode()}
+ $MEMBER_COLLECTOR_TERM = new
$CONVERT_COLLECTOR_TYPE_TERM(references);
+ }
+
+ @Override
+ public void open($STATE_DATA_VIEW_STORE store) throws Exception {
+ ${ctx.reuseOpenCode()}
+ }
+
+ @Override
+ public void accumulate($BASE_ROW $ACCUMULATE_INPUT_TERM) throws
Exception {
+ $accumulateCode
+ }
+
+ @Override
+ public void retract($BASE_ROW $RETRACT_INPUT_TERM) throws Exception {
+ $retractCode
+ }
+
+ @Override
+ public void merge($BASE_ROW $MERGED_ACC_TERM) throws Exception {
+ $mergeCode
+ }
+
+ @Override
+ public void setAccumulators($BASE_ROW $ACC_TERM) throws Exception {
+ $setAccumulatorsCode
+ }
+
+ @Override
+ public void resetAccumulators() throws Exception {
+ $resetAccumulatorsCode
+ }
+
+ @Override
+ public $BASE_ROW getAccumulators() throws Exception {
+ $getAccumulatorsCode
+ }
+
+ @Override
+ public $BASE_ROW createAccumulators() throws Exception {
+ $createAccumulatorsCode
+ }
+
+ @Override
+ public void emitValue(
+ $COLLECTOR<$BASE_ROW> $COLLECTOR_TERM, $BASE_ROW key, boolean
isRetract)
+ throws Exception {
+
+ $MEMBER_COLLECTOR_TERM.$COLLECTOR_TERM = $COLLECTOR_TERM;
Review comment:
can we add `COLLECTOR_TERM` to reset method?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
