[
https://issues.apache.org/jira/browse/FLINK-40075?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18110199#comment-18110199
]
dylanhz commented on FLINK-40075:
---------------------------------
We benchmarked a compatibility-safe PoC that keeps the original `STRUCTURED`
logical type while using `RowData` at runtime.
Against the legacy POJO accumulator, most results were within `-2.3%` to
`+2.4%` and were not stable across repeated isolated-JVM runs. Allocated
bytes/op were identical in every scenario.
Although this removes the POJO converter, the generated handler still copies
the state accumulator into a mutable `GenericRowData`, while `setField` may
introduce boxing. HotSpot can also optimize the simple three-field POJO
effectively, so no clear performance benefit remains.
Since we found neither a reproducible throughput improvement nor an allocation
reduction, we propose closing this issue for now. We can revisit it if the
runtime can avoid the remaining row copy and boxing.
> Optimize WELFORD_M2 aggregate accumulator by using RowData instead of
> STRUCTURED POJO accumulator
> -------------------------------------------------------------------------------------------------
>
> Key: FLINK-40075
> URL: https://issues.apache.org/jira/browse/FLINK-40075
> Project: Flink
> Issue Type: Improvement
> Components: Table SQL / Runtime
> Reporter: dylanhz
> Priority: Minor
>
> {{WelfordM2AggFunction}} is used for variance-related aggregates such as
> {{{}VAR_POP{}}}, {{{}VAR_SAMP{}}}, {{{}STDDEV_POP{}}}, and
> {{{}STDDEV_SAMP{}}}.
> Currently its accumulator is a structured POJO type. Using {{RowData}} as the
> accumulator could avoid unnecessary structured object conversion in generated
> aggregate code and improve performance on hot aggregate paths.
> This should be investigated as a separate optimization because changing the
> accumulator from {{STRUCTURED<WelfordM2Accumulator>}} to {{ROW<n, mean, m2>}}
> may affect compiled plan and streaming state/savepoint compatibility.
> *Validation*
> The implementation should include:
> * variance/stddev aggregate tests,
> * compiled plan restore coverage,
> * state/savepoint or serializer compatibility checks,
> * performance evidence for the reduced conversion overhead.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
