Yaoxuan Wu created FLINK-41179:
----------------------------------
Summary: 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
{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)
