jayzhan211 opened a new issue, #27221:
URL: https://github.com/apache/datafusion/issues/27221
### Describe the bug
When an aggregate groups by two columns with the same name from different
tables (`GROUP BY t1.a, t2.a`) and a partitioned hash join above it lists those
keys in another order, the query returns wrong rows. `EnsureRequirements`
reorders the aggregate's group keys to match the join, then restores the column
order with a projection that looks columns up by name, so both output columns
read the first `a`.
### To Reproduce
In `datafusion-cli`:
```sql
-- force partitioned hash joins on small tables
SET datafusion.optimizer.hash_join_single_partition_threshold = 0;
SET datafusion.optimizer.hash_join_single_partition_threshold_rows = 0;
CREATE TABLE t1 (a INT) AS VALUES (1), (2), (3);
CREATE TABLE t2 (a INT) AS VALUES (10), (20), (30);
CREATE TABLE t3 (k1 INT, k2 INT, v VARCHAR) AS VALUES (1, 20, 'x'), (2, 30,
'y'), (3, 10, 'z');
-- Query 1
SELECT g.a1, g.a2, t3.v
FROM (SELECT t1.a AS a1, t2.a AS a2 FROM t1 CROSS JOIN t2 GROUP BY t1.a,
t2.a) g
JOIN t3 ON g.a2 = t3.k2 AND g.a1 = t3.k1
ORDER BY t3.v;
-- Query 2
SELECT g.a1, g.a2, t3.v
FROM (SELECT t1.a AS a1, t2.a AS a2 FROM t1 CROSS JOIN t2 GROUP BY t1.a,
t2.a) g
JOIN t3 ON g.a1 = t3.k1 AND g.a2 = t3.k2
ORDER BY t3.v;
```
### Expected behavior
| Query | DataFusion returns | Expected |
|---|---|---|
| 1 | no rows | (1, 20, x), (2, 30, y), (3, 10, z) |
| 2 | (1, 20, x), (2, 30, y), (3, 10, z) | the same |
Query 1 also returns the right rows with `SET
datafusion.optimizer.top_down_join_key_reordering = false`.
### Additional context
<details>
<summary>EXPLAIN of query 1: both output columns read the first
<code>a</code> (<code>a@0 as a1, a@0 as a2</code>)</summary>
```text
HashJoinExec: mode=Partitioned, join_type=Inner, on=[(a2@1, k2@1), (a1@0,
k1@0)], projection=[a1@0, a2@1, v@4]
RepartitionExec: partitioning=Hash([a2@1, a1@0], 12), input_partitions=12
ProjectionExec: expr=[a@0 as a1, a@0 as a2]
AggregateExec: mode=FinalPartitioned, gby=[a@0 as a, a@1 as a], aggr=[]
RepartitionExec: partitioning=Hash([a@0, a@1], 12),
input_partitions=1
AggregateExec: mode=Partial, gby=[a@1 as a, a@0 as a], aggr=[]
CrossJoinExec
```
</details>
- Cause: `reorder_aggregate_keys` rebuilds the aggregate with the join's key
order, then builds the projection that restores the original order with
`agg_schema.index_of(name)`
(`datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs:376`).
The group columns are named without their table, so both are `a`, and
`index_of` returns the first one for both.
- This code dates from #7488 (2023).
- With default settings these small tables get a `CollectLeft` join, which
does not reorder keys. The two settings above force the partitioned join.
- Fix direction: build the restoring projection by position (where each
original key moved to), not by name.
### Willingness to contribute
I'm not planning to fix this myself, so anyone is welcome to pick it up.
Feel free to ping me with questions about the cause or when you'd like a review.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
