Wrong results: a list `unnest` keeps its input's GROUP BY key, so a later DISTINCT or GROUP BY drops rows

[adriangb]

Describe the bug

A list unnest turns one input row into many. But Unnest copies its input's functional dependencies without change. When the input is a GROUP BY (or a DISTINCT, or a table with a primary key), the optimizer still treats the grouping columns as a unique key after the unnest.

optimize_projections then removes the unnested column from a later GROUP BY or DISTINCT, because it seems to be determined by that key. The query returns one row per input key instead of one row per distinct value.

The bug occurs only when the parent of that aggregate does not use the removed column, for example count(*) over the DISTINCT. If the query selects the column, the optimizer keeps it and the result is correct. Thus the same DISTINCT gives the correct rows but an incorrect count(*).

To Reproduce

datafusion-cli:

CREATE TABLE t (k VARCHAR, v INT) AS VALUES ('a', 1), ('a', 2), ('b', 3);

-- 1. DISTINCT over the unnest output. Expected 3, actual 2.
WITH g AS (SELECT k, array_agg(v) AS vs FROM t GROUP BY k),
     u AS (SELECT k, unnest(vs) AS v FROM g)
SELECT count(*) AS n FROM (SELECT DISTINCT k, v FROM u);

-- 2. The same with GROUP BY. Expected 3, actual 2.
WITH g AS (SELECT k, array_agg(v) AS vs FROM t GROUP BY k),
     u AS (SELECT k, unnest(vs) AS v FROM g)
SELECT count(*) AS n FROM (SELECT k, v FROM u GROUP BY k, v);

-- 3. DISTINCT * directly over the unnest. Expected 3, actual 2.
WITH g AS (SELECT k, array_agg(v) AS vs FROM t GROUP BY k)
SELECT count(*) AS n FROM (SELECT DISTINCT * FROM (SELECT k, unnest(vs) FROM g));

-- 4. A list of structs, as returned by many aggregates and UDFs. Expected 3, actual 2.
WITH g AS (SELECT k, array_agg(named_struct('id', v)) AS rs FROM t GROUP BY k),
     u AS (SELECT k, unnest(rs) AS r FROM g)
SELECT count(*) AS n FROM (SELECT DISTINCT k, r['id'] FROM u);

-- 5. A window ORDER BY the former key. The default RANGE frame must give tied
--    rows the same running sum (3, 3, 6), so the expected result is 2. Actual 3:
--    the planner treats the ordering as strict and uses a ROWS frame (1, 3, 6).
WITH g AS (SELECT k, array_agg(v) AS vs FROM t GROUP BY k),
     u AS (SELECT k, unnest(vs) AS v FROM g)
SELECT count(DISTINCT s) AS n FROM (SELECT sum(v) OVER (ORDER BY k) AS s FROM u);

-- Control: the same DISTINCT without a GROUP BY below the unnest is correct (3).
SELECT count(*) AS n FROM (SELECT DISTINCT k, v FROM (SELECT k, unnest(make_array(v)) AS v FROM t));

The optimized plan of query 1 shows that v has gone from the DISTINCT aggregate:

Projection: count(Int64(1)) AS count(*) AS n
  Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]]
    Projection:
      Aggregate: groupBy=[[u.k]], aggr=[[]]            <-- expected groupBy=[[u.k, u.v]]
        SubqueryAlias: u
          Projection: g.k
            Unnest: lists[__unnest_placeholder(g.vs)|depth=1] structs[]
              Projection: g.k, g.vs AS __unnest_placeholder(g.vs)
                SubqueryAlias: g
                  Projection: t.k, array_agg(t.v) AS vs
                    Aggregate: groupBy=[[t.k]], aggr=[[array_agg(t.v)]]

EXPLAIN VERBOSE shows that optimize_projections removes v.

Expected behavior

Queries 1 to 4 return 3 and query 5 returns 2.

Versions

Version	Queries 1, 2, 3, 4	Query 5
54.0.0 (datafusion-cli release)	wrong	wrong
main at 991fd23 (2026-09-28)	wrong	wrong
#24787 head (f9ad1c3)	1, 2, 4 correct; 3 wrong	correct

Additional context

The cause is in Unnest::try_new (datafusion/expr/src/logical_plan/plan.rs):

// We can use the existing functional dependencies:
let deps = input_schema.functional_dependencies().clone();

The GROUP BY k below the unnest gives the dependency {k} -> all columns, with mode Single. After a list unnest, both parts of this dependency are wrong:

1. k is no longer unique. The mode must become Multi.
2. k does not determine the unnested column, which has the same index as the list column that it replaces. That index must be removed from the target set.

The draft PR #24787 found the same root cause through eliminate_join, and it corrects part 1. It does not correct part 2. Queries 1, 2 and 4 pass with #24787 only as a side effect: when a projection is above the unnest, project_functional_dependencies maps the targets of a Multi dependency column by column, and the aliased or derived column is not one of those targets. Query 3 has no projection between the unnest and the DISTINCT, so the unnested column is still a target. get_required_group_by_exprs_indices does not check the mode, and it removes the column.

A change that corrects both parts, in Unnest::try_new, passes all six queries above. It does this:

• Make every dependency Multi when a list column is unnested.
• Remove each dependency whose source includes an unnested list column.
• Remove the unnested list outputs from each target set.
• Map the indices from input to output positions through dependency_indices, because a struct unnest changes one input column into many output columns.

The unnest, functional_dependencies, group_by and window sqllogictest files also pass with this change. I can add these queries to the tests in #24787, or open a separate PR.