[
https://issues.apache.org/jira/browse/FLINK-40928?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18132568#comment-18132568
]
Yaoxuan Wu commented on FLINK-40928:
------------------------------------
Hi [~yusufyildiz34]!
Thanks for looking into this! Your analysis matches what I found. Happy to test
the fix once there is a PR.
> DISTINCT over a cross join with an IS NULL filter returns rows with NULL that
> are not in the input
> --------------------------------------------------------------------------------------------------
>
> Key: FLINK-40928
> URL: https://issues.apache.org/jira/browse/FLINK-40928
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 2.3.0
> Reporter: Yaoxuan Wu
> Priority: Major
>
> {code:java}
> CREATE TEMPORARY TABLE a (v INT) WITH ('connector' = 'filesystem', 'path' =
> 'file:///tmp/a', 'format' = 'csv');
> CREATE TEMPORARY TABLE x (v INT) WITH ('connector' = 'filesystem', 'path' =
> 'file:///tmp/x', 'format' = 'csv');
> INSERT INTO a VALUES (1), (2);
> INSERT INTO x VALUES (5); -- no NULL anywhereSELECT * FROM (SELECT
> DISTINCT a.v, x.v AS b FROM a, x) WHERE b IS NULL;
> -- Flink: (1, NULL), (2, NULL) expected: empty {code}
> The returned rows do not exist in the input: x contains only 5.
> * The wrong result only occurs when a and x are read from tables (here
> filesystem temporary tables). With the same rows written inline, FROM (VALUES
> (1), (2)) AS a(v), (VALUES (5)) AS x(v), the query correctly returns no rows.
> * Without the WHERE b IS NULL filter the result is correct: SELECT DISTINCT
> a.v, x.v AS b FROM a, x returns (1, 5), (2, 5). Removing DISTINCT also gives
> the correct (empty) result.
>
> In the optimized plan the DISTINCT is pushed below the join. On the x side
> the filter turns the grouping key into the constant NULL, the key is removed,
> and the grouped aggregate becomes a global aggregate, which returns one row
> even for an empty input:
> {code:java}
> Calc(select=[v, null:INTEGER AS b])
> +- NestedLoopJoin(joinType=[InnerJoin], where=[true], ...)
> :- HashAggregate(groupBy=[v], select=[v])
> +- Values(tuples=[[{ true }]], values=[dummy]) -- 2.0.1:
> HashAggregate(select=[]) {code}
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
