sepuri sai krishna created FLINK-41324:
------------------------------------------
Summary: COLLECT returns extra multiset entries when mini-batch is
enabled, and the result varies between runs
Key: FLINK-41324
URL: https://issues.apache.org/jira/browse/FLINK-41324
Project: Flink
Issue Type: Bug
Components: Table SQL / Runtime
Affects Versions: 2.4.0
Reporter: sepuri sai krishna
Enabling mini-batch changes the result of a query that applies {{COLLECT}} to
an updating input,
and the result is not the same from one run to the next. Mini-batch is
documented as an
optimization that trades latency for throughput, so it should not change
results at all.
{code:sql}
-- t holds (1,1.0) (1,2.0) (1,4.0) (2,10.0) (2,20.0) (3,7.0) (3,1.0)
SELECT CAST(COLLECT(s) AS STRING)
FROM (SELECT k, SUM(x) AS s FROM t GROUP BY k);
{code}
The inner query produces 7.0, 30.0 and 8.0, so the answer is {{{7.0=1, 30.0=1,
8.0=1}}}. That is
what comes back without mini-batch, and in batch mode.
With {{table.exec.mini-batch.enabled}} set to true the result gains entries
whose multiplicity is
0, and it differs between runs of the same query on the same data:
{noformat}
{8.0=1, 10.0=0, 3.0=0, 7.0=1, 30.0=1}
{1.0=0, 8.0=1, 7.0=1, 30.0=1}
{8.0=1, 7.0=1, 30.0=1} correct, on some runs
{noformat}
A multiset entry with multiplicity 0 means the element is not in the multiset,
so it should not
appear. The extra values are intermediate results of the inner aggregate that
no longer exist.
Config: parallelism 1, {{table.exec.mini-batch.enabled=true}},
{{allow-latency=10 ms}},
{{size=2}}, and the default {{table.optimizer.agg-phase-strategy}}. Also
reproduced with
{{size=1000}} on some runs, and with {{allow-latency=1 s}}.
Scope: {{UNNEST}} of the result is not affected, it returns three rows either
way. The wrong value
is in the multiset itself.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
