[
https://issues.apache.org/jira/browse/FLINK-41037?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18132185#comment-18132185
]
Yusuf YILDIZ commented on FLINK-41037:
--------------------------------------
Hi, I'd like to work on this. The root cause is the fallback in
FlinkRelMdUniqueGroups#getUniqueGroups(Join), which unions the unique groups of
both inputs regardless of the join type. On the null-padded side of an outer
join,
padded rows fall into the same group as real rows whose unique group columns are
NULL, so AggregateReduceGroupingRule creates invalid auxiliary groups.
Proposed fix: on a null-padded input, keep the reduced groups only if they
contain
a column that is never NULL on the real rows (a NOT NULL column, or a key of a
null-rejecting equi-condition when that input is not preserved by the join).
Only batch plans are affected, since the rule is batch-only. I have a fix with
tests ready. Could a committer please assign this ticket to me?
> DISTINCT over a FULL OUTER JOIN drops a row when the grouping key is NULL
> (batch mode)
> --------------------------------------------------------------------------------------
>
> Key: FLINK-41037
> URL: https://issues.apache.org/jira/browse/FLINK-41037
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 2.3.0
> Reporter: Yaoxuan Wu
> Priority: Major
>
> {code:python}
> from pyflink.table import EnvironmentSettings, TableEnvironment, DataTypes as
> D
> t = TableEnvironment.create(EnvironmentSettings.in_batch_mode())
> t.create_temporary_view('l', t.from_elements([(None, 1.0)],
> D.ROW([D.FIELD('k', D.STRING()), D.FIELD('x', D.DOUBLE())])))
> t.create_temporary_view('r', t.from_elements([(None,)], D.ROW([D.FIELD('k',
> D.STRING())])))
> J = "SELECT g.k, g.s, r.k AS rk FROM (SELECT k, SUM(x) AS s FROM l GROUP BY
> k) g FULL JOIN r ON g.k = r.k"
> t.execute_sql(J).print()
> # (NULL, 1.0, NULL), (NULL, NULL, NULL) correct: two distinct rows
> t.execute_sql(f"SELECT DISTINCT * FROM ({J})").print()
> # expected: (NULL, 1.0, NULL), (NULL, NULL, NULL)
> # actual: (NULL, NULL, NULL)
> {code}
> The plan groups by {{k, rk}} and keeps {{s}} as {{auxGrouping}}, i.e. it
> assumes {{k}} determines {{s}}. That holds for the GROUP BY output but not
> after a FULL or RIGHT join, which pads the left side with NULLs: the real row
> (NULL, 1.0) and the padded row (NULL, NULL) fall into one group. The
> assumption comes from {{FlinkRelMdUniqueGroups.getUniqueGroups}}, whose
> fallback for a Join ignores the join type. Streaming mode is correct.
> Reproduces on 2.0.1, 2.2.1, 2.3.0 and master.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
