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_r310109672
##########
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;
+ $MEMBER_COLLECTOR_TERM.reset(key, isRetract);
+ $emitValueCode
+ }
+
+ @Override
+ public void cleanup() throws Exception {
+ ${ctx.reuseCleanupCode()}
+ }
+
+ @Override
+ public void close() throws Exception {
+ ${ctx.reuseCloseCode()}
+ }
+
+ private class $CONVERT_COLLECTOR_TYPE_TERM implements $COLLECTOR {
+ public $COLLECTOR<$BASE_ROW> $COLLECTOR_TERM;
+ private $BASE_ROW key;
+ private $JOINED_ROW result;
+ private boolean isRetract = false;
+ private static final byte ACCUMULATE_MSG = 0;
Review comment:
use `BaseRowUtil.ACCUMULATE_MSG` instead of hard code?
----------------------------------------------------------------
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
