[
https://issues.apache.org/jira/browse/FLINK-41200?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-41200:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Priority: Major
> Labels: pull-request-available
>
> 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)
