Yaoxuan Wu created FLINK-41098:
----------------------------------
Summary: Streaming mode: IN / NOT IN in the SELECT list drops all
rows when the subquery result is empty
Key: FLINK-41098
URL: https://issues.apache.org/jira/browse/FLINK-41098
Project: Flink
Issue Type: Bug
Components: Table SQL / Planner, Table SQL / Runtime
Affects Versions: 2.3.0, 2.0.1
Reporter: Yaoxuan Wu
{code:python}
from pyflink.table import EnvironmentSettings, TableEnvironment, DataTypes as D
t = TableEnvironment.create(EnvironmentSettings.in_streaming_mode())
t.create_temporary_view('t', t.from_elements([(1,), (2,)], D.ROW([D.FIELD('k',
D.INT())])))
t.execute_sql("SELECT k, k NOT IN (SELECT k FROM t WHERE k > 99) AS x FROM
t").print()
# expected: (1, TRUE), (2, TRUE) (batch mode returns this)
# actual: no rows
t.execute_sql("SELECT k, (SELECT COUNT(*) FROM t WHERE k > 99) AS c FROM
t").print()
# expected: (1, 0), (2, 0)
# actual: (1, NULL), (2, NULL)
{code}
The same for {{IN}}. The input is bounded and the subquery is empty, so every
outer row must be kept; batch mode is correct.
Cause: {{SubQueryRemoveRule}} rewrites {{k NOT IN (subquery)}} in a SELECT list
into an inner join of the outer rows with a keyless {{COUNT(*), COUNT(k)}}
aggregate over the subquery (plus a left join for the matches), and a scalar
{{COUNT(*)}} subquery into a left join with such an aggregate. A keyless
aggregate must return one row even for an empty input (COUNT = 0). The batch
operator does this ({{AggWithoutKeysCodeGenerator}} emits the initial values at
end of input if there was no input), but the streaming {{GroupAggFunction}}
only emits from {{processElement}}, so over an empty input it emits nothing,
and the inner join drops every outer row:
{code}
Join(LeftOuterJoin, where=[(k = k0)])
:- Join(InnerJoin, where=[true])
: :- TableSourceScan(fields=[k])
: +- GroupAggregate(select=[COUNT(*) AS c, COUNT(k) AS ck]) -- no row for an
empty input
: +- Calc(select=[k], where=[(k > 99)])
+- ...
{code}
Reproduces on 2.0.1, 2.2.1, 2.3.0 and master.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
