yusufyildiz34 opened a new pull request, #29666:
URL: https://github.com/apache/flink/pull/29666
## What is the purpose of the change
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. Take `l(k, x) = (4, 20), (4, 1)` and `r(id, y,
z) = (4, 2, 'Z')`, where `r` is unique on `id`. The query `SELECT k, y, z,
COUNT(*) FROM l LEFT JOIN r ON k = id AND x > y GROUP BY k, y, z` merges both
rows into one group with `COUNT(*) = 2` instead of returning `(4, 2, Z, 1), (4,
NULL, NULL, 1)`.
`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. It assumes that these groups then determine the columns
of the other input. But a row of the preserved input is padded with nulls if
the non-equi part of the join condition fails for it. Rows with the same unique
group values can therefore get different columns of the other input, and
`AggregateReduceGroupingRule` turns those columns into invalid auxiliary
groups. `AggregateReduceGroupingRule` is batch-only, so streaming plans are not
affected.
## Brief change log
- For LEFT and RIGHT joins, `FlinkRelMdUniqueGroups` now only takes this
shortcut in two cases:
- the unique groups of the preserved input are unique in that input, so
they identify a single row;
- the non-equi part of the join condition is deterministic and only uses
requested columns of the preserved input.
- Plans of equi-joins don't change, because their non-equi part is empty.
- The JIRA suggests requiring an equi-condition. That is more conservative
than needed: it would also give up the optimization in the cases above.
This PR is independent of #29665 (FLINK-41037), which fixes another issue in
the same method. Whichever is merged second needs a trivial rebase of the test
file.
## Verifying this change
This change added tests and can be verified as follows:
- `AggregateReduceGroupingITCase#testAggOnLeftJoinWithNonEquiCondition` and
`#testAggOnRightJoinWithNonEquiCondition` reproduce the issue for both join
directions. Without the fix they return 4 instead of 5 rows.
- `AggregateReduceGroupingTest` /
`AggregateReduceGroupingRuleTest#testAggOnLeftJoinWithNonEquiCondition`: the
plan no longer uses auxiliary groups.
-
`FlinkRelMdUniqueGroupsTest#testGetUniqueGroupsOnOuterJoinWithNonEquiCondition`:
the groups are not reduced when the non-equi condition uses a column of the
preserved input that is not requested, or when it is non-deterministic. They
are still reduced when it only uses requested columns, or when the groups are
unique in the preserved input.
- This commit changes no existing test expectation or golden plan. All
planner plan tests (`org.apache.flink.table.planner.plan.**`, excluding restore
and semantic tests) and the batch aggregate / join ITCases pass.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Claude Opus 5.5)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
