Yusuf YILDIZ created FLINK-41200:
------------------------------------
Summary: GROUP BY over an outer join with a non-equi join
condition drops rows (batch mode)
Key: FLINK-41200
URL: https://issues.apache.org/jira/browse/FLINK-41200
Project: Flink
Issue Type: Bug
Components: Table SQL / Planner
Reporter: Yusuf YILDIZ
In batch mode, a GROUP BY over a LEFT or RIGHT join returns wrong results when
the join condition has a non-equi predicate in addition to an equi-condition on
a unique key.
{code:sql}
CREATE TABLE l (k INT, x INT) WITH ('connector' = 'filesystem', 'path' =
'file:///tmp/l', 'format' = 'csv');
CREATE TABLE r (id INT, y INT, z STRING) WITH ('connector' = 'filesystem',
'path' = 'file:///tmp/r', 'format' = 'csv');
INSERT INTO l VALUES (4, 20), (4, 1);
INSERT INTO r VALUES (4, 2, 'Z');
SELECT k, y, z, COUNT(*) FROM l LEFT JOIN
(SELECT id, MAX(y) AS y, MAX(z) AS z FROM r GROUP BY id) ON k = id AND x > y
GROUP BY k, y, z;
-- expected: (4, 2, Z, 1), (4, NULL, NULL, 1)
-- actual: (4, 2, Z, 2)
{code}
The row (4, 20) matches r, the row (4, 1) does not (1 > 2 is false) and is
padded with
NULLs. The optimized plan groups by k only and keeps y and z as auxiliary
groups:
{noformat}
SortAggregate(isMerge=[false], groupBy=[k], auxGrouping=[y, z], select=[k, y,
z, COUNT(*) AS EXPR$3])
+- Calc(select=[k, y, z])
+- SortMergeJoin(joinType=[LeftOuterJoin], where=[AND(=(k, id), >(x, y))],
...)
{noformat}
{*}Cause{*}: FlinkRelMdUniqueGroups.getUniqueGroups(Join) returns the unique
groups of the preserved input if they contain its join keys and the join keys
of the other input are
unique. That only holds for a pure equi-condition: with a non-equi predicate,
two rows with the same join key can match and not match the same row, so the
join key does not determine the columns of the other input.
AggregateReduceGroupingRule then turns those columns into auxiliary groups. The
same happens for a RIGHT JOIN with the inputs swapped. Streaming mode is not
affected, as AggregateReduceGroupingRule is batch-only.
{*}Possible fix{*}: in both branches, also require the join condition to be an
equi-condition for outer joins. Reproduces on master (2.4-SNAPSHOT). Related to
FLINK-41037 (same method).
I'd like to work on this and have reproducing tests ready.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
