[
https://issues.apache.org/jira/browse/FLINK-41179?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18132175#comment-18132175
]
dylanhz commented on FLINK-41179:
---------------------------------
Hi [~thaddywu] , thanks for reporting this issue! Are you currently working on
a fix? If not, I’d be happy to take it on.
> 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)
