[
https://issues.apache.org/jira/browse/FLINK-41249?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18133347#comment-18133347
]
Andrew Stein commented on FLINK-41249:
--------------------------------------
We hit the same bug on 2.1.1, in a SQL job with a chain of ten left outer joins
on ForSt where rows showed up under the wrong keys, so 2.1.x is affected as
well. The code is identical in every release from 2.0.0 through 2.3.0.
This adds two details to the report.
*The outer-join view also corrupts state.* In a left outer join, when a
right-side record picks up another key's left row from the shared list, the
operator emits a retraction and a joined row under that other key, and
{{updateNumOfAssociations}} then writes the other key's row into the current
key's left state. The wrong row is checkpointed and keeps producing output
under the wrong key until the current key's own left record is updated again.
Because the results are silently wrong and survive restarts, this may deserve
Critical rather than Major.
*Why the operator tests miss it.*
{{AsyncKeyedTwoInputStreamOperatorTestHarness}} defaults to
{{HashMapStateBackend}}, whose async adaptor completes state futures
synchronously, so the stages never interleave. A harness test that sets
{{ForStStateBackend}} reproduces it reliably: a left outer join with both
inputs unique on the join key, 1,000 left rows, a drain, then the 1,000
matching right rows and a drain. On 2.1.1, 999 of the 1,000 joined rows paired
different keys, and all of the right rows landed on 20 left rows. Allocating
the list per call in both {{JoinKeyContainsUniqueKey}} views fixes it, and the
existing {{StreamingJoinOperatorTest}} and
{{StreamingSemiAntiJoinOperatorTest}} still pass.
I have that fix and the ForSt harness test ready and can open a pull request if
no one is working on it yet.
> Async-state streaming join returns rows joined with another key's match when
> the join key contains the unique key (ForSt)
> -------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41249
> URL: https://issues.apache.org/jira/browse/FLINK-41249
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
> Affects Versions: 2.0.1, 2.3.0, 2.2.1
> Reporter: Yaoxuan Wu
> Priority: Major
>
> {code:python}
> from pyflink.common import Configuration
> from pyflink.table import EnvironmentSettings, TableEnvironment, DataTypes as
> D
> c = Configuration()
> c.set_string('parallelism.default', '1')
> c.set_string('table.exec.async-state.enabled', 'true')
> c.set_string('state.backend.type', 'forst')
> t =
> TableEnvironment.create(EnvironmentSettings.new_instance().in_streaming_mode().with_configuration(c).build())
> t.create_temporary_view('t', t.from_elements([(i, 'v%d' % i) for i in
> range(7)],
> D.ROW([D.FIELD('k', D.INT()), D.FIELD('s', D.STRING())])))
> rows = t.execute_sql("""SELECT a.k FROM (SELECT DISTINCT k FROM t) a
> JOIN (SELECT DISTINCT k FROM t) b ON a.k =
> b.k""").collect()
> print(sorted(r[0] for r in rows))
> # expected: [0, 1, 2, 3, 4, 5, 6]
> # actual (varies between runs): [6, 6, 6, 6, 6, 6, 6], [0, 0, 0, 0, 0, 0, 0],
> ...
> {code}
> Correct with synchronous state (on hashmap or ForSt) and with async state on
> the hashmap backend; wrong in every run with async state on ForSt.
> Cause: when the join key contains the input's unique key,
> {{JoinRecordAsyncStateViews.JoinKeyContainsUniqueKey}} keeps one
> {{List<OuterRecord> reusedList}} per view and returns it from every
> {{findMatchedRecords}} call ({{reusedList.clear(); reusedList.add(...);
> return reusedList;}}). {{AssociatedRecords.fromAsyncStateView}} wraps that
> list lazily ({{.thenApply(AssociatedRecords::new)}}), and
> {{AsyncStateStreamingJoinOperator.doProcessElement}} iterates it later in
> {{.thenAccept(... joinHelper.processJoin(...))}}. With a truly asynchronous
> backend (ForSt) these stages of different in-flight records interleave, so
> another record's lookup clears and refills the list before the first record's
> {{processJoin}} reads it: the record is joined with the other key's match
> (its join condition was never checked against it).
> {{OuterJoinRecordAsyncStateViews.JoinKeyContainsUniqueKey}} has the same
> shared {{reusedList}}. The synchronous views are safe because each record
> finishes before the next starts. Allocating the (at most one-element) list
> per call would fix it.
> Reproduces on 2.0.1, 2.2.1 and 2.3.0; master has the same code.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
