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