[
https://issues.apache.org/jira/browse/FLINK-41179?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18132567#comment-18132567
]
Yaoxuan Wu commented on FLINK-41179:
------------------------------------
[~dylanhz] Thanks for picking this up! I'm not working on a fix, please go
ahead.
In case it helps: with mini-batch two-phase aggregation,
MiniBatchLocalGroupAggFunction starts each bundle from a fresh accumulator, so
a bundle that begins with -U takes the local count to -1.
WelfordM2AggFunction.accumulate / retract return early while acc.n <= 0
(dropping the value), and merge only adds the count of a partial accumulator
with n <= 0, so the retracted and the new value are both lost. Happy to test
the fix.
> VAR_POP / STDDEV_POP give wrong, non-deterministic results with mini-batch
> two-phase aggregation over an updating input
> -----------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41179
> URL: https://issues.apache.org/jira/browse/FLINK-41179
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
> Affects Versions: 2.3.0
> Reporter: Yaoxuan Wu
> Priority: Major
>
> {code:python}
> from collections import Counter
> from pyflink.common import Configuration
> from pyflink.table import DataTypes as D, EnvironmentSettings,
> TableEnvironment
> c = Configuration()
> for k, v in [('parallelism.default', '1'), ('table.exec.mini-batch.enabled',
> 'true'),
> ('table.exec.mini-batch.allow-latency', '10 ms'),
> ('table.exec.mini-batch.size', '2'),
> ('table.optimizer.agg-phase-strategy', 'TWO_PHASE')]:
> c.set_string(k, v)
> t =
> TableEnvironment.create(EnvironmentSettings.new_instance().in_streaming_mode().with_configuration(c).build())
> t.create_temporary_view('t', t.from_elements(
> [(1, 1.0), (1, 2.0), (1, 4.0), (2, 10.0), (2, 20.0), (3, 7.0), (3, 1.0)],
> D.ROW([D.FIELD('k', D.INT()), D.FIELD('x', D.DOUBLE())])))
> final = Counter() # fold the changelog into the final result
> with t.execute_sql("SELECT VAR_POP(s), AVG(s) FROM (SELECT k, SUM(x) AS s
> FROM t GROUP BY k)").collect() as it:
> for r in it:
> final[tuple(r)] += 1 if str(r.get_row_kind()) in ('+I', '+U') else -1
> print([k for k, n in final.items() if n > 0])
> # SUM per k = 7, 30, 8
> # expected: [(112.666..., 15.0)]
> # actual (varies between runs): [(130.888..., 15.0)], [(9.555..., 15.0)], ...
> {code}
> AVG / SUM / COUNT of the same input are correct; VAR_POP, VAR_SAMP,
> STDDEV_POP and STDDEV_SAMP are wrong. Correct without mini-batch, with
> {{ONE_PHASE}}, in batch mode, and on 2.2.1.
> Cause: since FLINK-38740 the variance functions use {{WelfordM2AggFunction}}.
> With mini-batch two-phase aggregation, {{MiniBatchLocalGroupAggFunction}}
> starts each bundle from a fresh accumulator, so a bundle with {{-U(s1),
> +U(s2)}} first takes the local count to -1. But
> {{WelfordM2AggFunction.retract}} and {{accumulate}} return early when {{acc.n
> <= 0}} (dropping the value), and {{merge}} only adds the count of an
> accumulator with {{n <= 0}}. So the global accumulator still has s1 in mean /
> m2 and never sees s2.
> Reproduces on 2.3.0 and master; 2.2.1 is correct.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
