Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/4355#discussion_r130117980
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
---
@@ -108,30 +112,38 @@ object AggregateUtil {
outputArity,
needRetract = false,
needMerge = false,
- needReset = false
+ needReset = false,
+ accConfig = Some(DataViewConfig(accSpecs, isUseState))
)
+ val accConfig = accSpecs
+ .flatMap(specs => specs.map(s => (s.id, s.toStateDescriptor)))
+ .toMap[String, StateDescriptor[_, _]]
+
if (isRowTimeType) {
if (isRowsClause) {
// ROWS unbounded over process function
new RowTimeUnboundedRowsOver(
genFunction,
aggregationStateType,
CRowTypeInfo(inputTypeInfo),
- queryConfig)
+ queryConfig,
+ accConfig)
--- End diff --
I think it would be better to keep `accConfig` out of the ProcessFunctions.
It is just passing information to the `GeneratedAggregations` which could also
be code generated. The only thing that we need is a `RuntimeContext` in
`GeneratedAggregations`. Therefore, I propose to add a method
`GeneratedAggregations.initialize(ctx: RuntimeContext())` instead of adding
`GeneratedAggregations.setDataViewFactory()`. In `initialize()` we can generate
code that registers all necessary state by itself and keeps it as member
variables.
I think this would be cleaner because it encapsulates everything that's
related to aggregation functions in the code-gen'd class.
If we use heap state in `MapView` and `ListView` as default, we also won't
need `DataViewFactory` because we can generate all state access directly (if
required).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
