yusufyildiz34 opened a new pull request, #29664:
URL: https://github.com/apache/flink/pull/29664
## What is the purpose of the change
`FlinkAggregateJoinTransposeRule` pushes an aggregate below a join. For a
join input that contributes no grouping or join key, it creates an aggregate
without grouping keys on that input, which returns a row even for an empty
input. The rewritten plan can then return rows, or wrong aggregate values,
where the original join is empty:
- A DISTINCT over a cross join with an input that is empty after filtering
returns rows. For the query from the JIRA, `SELECT * FROM (SELECT DISTINCT a.v,
x.v AS b FROM a, x) WHERE b IS NULL` returns `(1, NULL), (2, NULL)` instead of
no rows.
- Over a cross join with an empty input, SUM returns `0` and MIN returns a
value instead of `NULL`.
- A global COUNT(*) over a cross join with an empty single-row input (e.g. a
`HAVING` subquery) returns no row instead of `0`.
- A DISTINCT pushed to the input without keys even becomes `VALUES(true)`,
which drops that input and its filter from the plan. This also gives wrong
results in streaming mode.
## Brief change log
- `FlinkAggregateJoinTransposeRule` only pushes an aggregate without
grouping keys to a join input if the original aggregate has no grouping keys
either, only computes COUNT / $SUM0, and both inputs are aggregated. Each input
then returns exactly one row, so the result for an empty input stays correct.
COUNT(*) over a cross join is still computed as a product of counts, so
existing plans of this shape (e.g. in `JoinHintTestBase`) do not change.
- Mention the change in the class Javadoc that lists the modifications to
Calcite's `AggregateJoinTransposeRule`.
- This changes the optimized plan of affected queries, including streaming
ones. It therefore targets master only and should not be backported to patch
releases.
Calcite's `AggregateJoinTransposeRule` (1.41) has no such guard either, so
it seems to have the same issue.
## Open question
In streaming mode, an aggregate without grouping keys emits nothing for an
empty input. Grouped aggregates with calls over a cross join (e.g. `SELECT a,
COUNT(*) FROM A, B GROUP BY a`) were therefore already correct there, and this
change no longer pushes their counts down. A more general alternative would
keep the push-down in both modes: add a `COUNT(*) > 0` filter on the pushed
aggregate without keys, and keep the top aggregate for global aggregates. I
kept the simpler fix to limit the risk, and I'm happy to implement the
alternative if you prefer.
## Verifying this change
This change added tests and can be verified as follows:
- `JoinWithoutKeyITCase#testDistinctOnCrossJoinWithIsNullFilter` runs an
adapted version of the JIRA query (`Table3` × `SmallTable3`) with the default
optimizer. Without the fix it returns 21 rows instead of none.
- `AggregateJoinTransposeITCase#testAggOverCrossJoinWithEmptyInput` covers:
- DISTINCT with the input without keys on either side
- GROUP BY with COUNT(*)
- global COUNT / SUM / MIN, and COUNT only
- an empty single-row (`HAVING`) input
- `FlinkAggregateJoinTransposeRuleTest` covers plans for:
- a grouped aggregate, a global SUM / MIN and a global COUNT(*) over an
empty single-row input, which are not pushed
- a global COUNT(*) over a cross join, which is still pushed
- `JoinTest#testDistinctOnCrossJoinWithIsNullFilter` (streaming): the plan
keeps the filtered input instead of `VALUES(true)`.
- No existing plan output changed; the golden files only gained the new
cases. All planner plan tests and 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]
