Join dynamic filter pushdown through FilterExec with a projection can discard matching rows with duplicate column names #25262

SET datafusion.optimizer.join_reordering = false;

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

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

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

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

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

SELECT s.id AS sid, a.id AS aid, b.id AS bid
FROM tb s
JOIN (
    SELECT a.id, b.id
    FROM ta a LEFT JOIN tb b ON a.ty = b.id
    WHERE (a.ty || '!') IS DISTINCT FROM b.id
)
ON s.id = b.id;

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

SELECT s.id AS sid, a.id AS aid, b.id AS bid
FROM tb s
JOIN (
    SELECT a.id, b.id
    FROM ta a LEFT JOIN tb b ON a.ty = b.id
    WHERE (a.ty || '!') IS DISTINCT FROM b.id
)
ON s.id = b.id;