[
https://issues.apache.org/jira/browse/FLINK-41249?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Yaoxuan Wu updated FLINK-41249:
-------------------------------
Affects Version/s: 2.2.1
2.3.0
2.0.1
> 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)
