[
https://issues.apache.org/jira/browse/FLINK-40928?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18132211#comment-18132211
]
Yusuf YILDIZ edited comment on FLINK-40928 at 10/26/26, 7:22 AM:
-----------------------------------------------------------------
Hi, I'd like to work on this. The root cause is in
FlinkAggregateJoinTransposeRule:
when a join input contributes no grouping or join key, the rule pushes an
aggregate
without grouping keys to it, which returns a row even for an empty input.
Besides
the DISTINCT case, this also makes e.g. SUM/MIN over a cross join with an empty
input return 0 / a value instead of NULL, and in streaming mode the empty input
is
replaced by VALUES(true).
Proposed fix: only push an aggregate without grouping keys if the original
aggregate has no grouping keys either, only computes COUNT/SUM0, and both inputs
are aggregated (so COUNT( * ) over a cross join is still computed as a product
of
counts). I have a fix with tests ready. Could a committer please assign this
ticket to me?
was (Author: JIRAUSER315263):
Hi, I'd like to work on this. The root cause is in
FlinkAggregateJoinTransposeRule:
when a join input contributes no grouping or join key, the rule pushes an
aggregate
without grouping keys to it, which returns a row even for an empty input.
Besides
the DISTINCT case, this also makes e.g. SUM/MIN over a cross join with an empty
input return 0 / a value instead of NULL, and in streaming mode the empty input
is
replaced by VALUES(true).
Proposed fix: only push an aggregate without grouping keys if the original
aggregate has no grouping keys either, only computes COUNT/SUM0, and both inputs
are aggregated (so COUNT(*) over a cross join is still computed as a product of
counts). I have a fix with tests ready. Could a committer please assign this
ticket to me?
> 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)
