Dynamic filters sometimes do not get pushed down through aggregations

[jayshrivastava]

Describe the bug

Consider this plan

  HashJoinExec: on=[(bhandle@0, bhandle@0)]       ← creates dynamic filter on bhandle@0
    DataSourceExec (build)
    AggregateExec: gby=[bhandle@1], aggr=[min]     ← output: bhandle@0, min@1
      ProjectionExec: [value@1, bhandle@0]          ← reorders columns
        DataSourceExec (probe)                      ← filter should be pushed down here but it's not

The dynamic filter does not get pushed down through the AggregateExec due to a mixup between input and output column indexes.

The issue happens due to the ProjectionExec reordering columns.

To Reproduce

Tested on df 52.1.0

  CREATE TABLE contexts (a VARCHAR)
    AS VALUES ('h1'), ('h2');

  CREATE TABLE data (a VARCHAR, ts TIMESTAMP, value DOUBLE)
    AS VALUES
      ('h1', '2024-01-01T00:05:00', 1.0),
      ('h1', '2024-01-01T00:15:00', 2.0),
      ('h2', '2024-01-01T00:25:00', 3.0),
      ('h3', '2024-01-01T00:35:00', 4.0);

EXPLAIN VERBOSE SELECT * FROM contexts c
  INNER JOIN (
    SELECT a, date_bin(interval '1 hour', ts) AS bucket, min(value) AS min_val
    FROM (SELECT value, a, ts FROM data)
    GROUP BY a, date_bin(interval '1 hour', ts)
  ) agg ON c.a = agg.a;

The output plan does not push down dynamic filters

HashJoinExec: mode=Auto, join_type=Inner, on=[(a@0, a@0)]
      DataSourceExec: partitions=1
      ProjectionExec: [a@0, date_bin(1h, ts)@1 as bucket, min(value)@2 as min_val]
        AggregateExec: mode=FinalPartitioned, gby=[a@0, date_bin(1h, ts)@1], aggr=[min(value)]
          AggregateExec: mode=Partial, gby=[a@1, date_bin(1h, ts@2)], aggr=[min(value)]
            ProjectionExec: [value@2, a@0, ts@1]        ← reorders columns
              DataSourceExec: partitions=1

Expected behavior

No response

Additional context

No response

[jayshrivastava]

take