Join dynamic filter pushdown through AggregateExec can confuse same-named grouping columns #25264

SET datafusion.optimizer.join_reordering = false;

COPY (SELECT 'a1' AS id, 'x1' AS ty)
TO '/tmp/df_dynamic_filter_aggregate_a.parquet'
STORED AS PARQUET;

COPY (SELECT 'x1' AS id)
TO '/tmp/df_dynamic_filter_aggregate_b.parquet'
STORED AS PARQUET;

CREATE EXTERNAL TABLE ta
STORED AS PARQUET
LOCATION '/tmp/df_dynamic_filter_aggregate_a.parquet';

CREATE EXTERNAL TABLE tb
STORED AS PARQUET
LOCATION '/tmp/df_dynamic_filter_aggregate_b.parquet';

-- Returns one row: (x1, a1, x1).
SET datafusion.optimizer.enable_join_dynamic_filter_pushdown = false;

SELECT s.id, a.id, b.id
FROM tb s
JOIN (
    SELECT a.id, b.id
    FROM ta a LEFT JOIN tb b ON a.ty = b.id
    GROUP BY a.id, b.id
)
ON s.id = b.id;

-- Incorrectly returns zero rows.
SET datafusion.optimizer.enable_join_dynamic_filter_pushdown = true;

SELECT s.id, a.id, b.id
FROM tb s
JOIN (
    SELECT a.id, b.id
    FROM ta a LEFT JOIN tb b ON a.ty = b.id
    GROUP BY a.id, b.id
)
ON s.id = b.id;