[
https://issues.apache.org/jira/browse/FLINK-41324?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18137945#comment-18137945
]
sepuri sai krishna commented on FLINK-41324:
--------------------------------------------
This may share a trigger with FLINK-41179.
That ticket reports {{VAR_POP}} and {{STDDEV_POP}} going wrong under
mini-batch, and its repro uses
the same seven input rows and the same mini-batch settings as the description
above. The
explanation posted there is that under two-phase aggregation the local
aggregate starts each bundle
from a fresh accumulator, so a bundle that opens with a retraction retracts a
value the accumulator
has never seen.
That fits what I measured here. The result is correct with {{ONE_PHASE}},
correct with mini-batch
off, and correct in batch mode, which is the same pattern that ticket reports.
Where the two differ is what the aggregate then does with it. There the value
is dropped. Here the
multiset keeps an entry whose multiplicity is 0, so the same starting point
produces two different
wrong answers.
Whether this needs a separate fix depends on where the fix goes. At the
mini-batch local
accumulator it would cover both. Per aggregate function it would not.
Raising it so the two are not treated as unrelated, and so this one is not
closed as a duplicate
before that is decided.
> 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
> Priority: Major
>
> 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)
