SELECT DISTINCT ... LIMIT n with a window function fails with Internal error: Ordering direction required for DISTINCT with limit #25350

Describe the bug

SELECT DISTINCT ... LIMIT n with no ORDER BY fails when one of the projected columns is a window function. Planning succeeds. Execution fails with:

Internal error: Ordering direction required for DISTINCT with limit.
This issue was likely caused by a bug in DataFusion's code. Please help us to resolve this by filing a bug report in our issue tracker: https://github.com/apache/datafusion/issues

To Reproduce

datafusion-cli 55.1.0, built from main at 8bd6629:

CREATE TABLE t (env VARCHAR, region VARCHAR, deployment VARCHAR);
INSERT INTO t VALUES ('prod','a','x'), ('prod','a','y'), ('staging','b','x');

-- Fails: Internal error: Ordering direction required for DISTINCT with limit.
SELECT DISTINCT env, region, count(*) OVER (PARTITION BY env, deployment) AS n FROM t LIMIT 30;

-- Also fails
SELECT DISTINCT env, count(*) OVER (PARTITION BY env) AS n FROM t LIMIT 30;

-- Works (no window function)
SELECT DISTINCT env, region FROM t LIMIT 30;

The failure also occurs with set datafusion.execution.target_partitions = 1;.

The same query succeeds and returns the correct rows in each of these variants:

-- no LIMIT
SELECT DISTINCT env, region, count(*) OVER (PARTITION BY env, deployment) AS n FROM t;

-- ORDER BY before the LIMIT
SELECT DISTINCT env, region, count(*) OVER (PARTITION BY env, deployment) AS n FROM t ORDER BY env LIMIT 30;

-- soft limit pushdown disabled
set datafusion.optimizer.enable_distinct_aggregation_soft_limit = false;
SELECT DISTINCT env, region, count(*) OVER (PARTITION BY env, deployment) AS n FROM t LIMIT 30;

Expected behavior

The query returns the distinct rows, as it does without the LIMIT. PostgreSQL 16.15 and DuckDB 1.5.2 return:

env     | region | n
--------+--------+---
prod    | a      | 1
staging | b      | 1

and, for the single-column form:

env     | n
--------+---
prod    | 2
staging | 1

Additional context

EXPLAIN FORMAT INDENT (abbreviated) shows that each AggregateExec has a pushed limit (lim=[30]) and also an ordering mode (ordering_mode=PartiallySorted([0])):

CoalescePartitionsExec: fetch=30
  AggregateExec: mode=FinalPartitioned, gby=[env@0 as env, region@1 as region, n@2 as n], aggr=[], lim=[30], ordering_mode=PartiallySorted([0])
    RepartitionExec: partitioning=Hash([env@0, region@1, n@2], 12), input_partitions=1, maintains_sort_order=true
      AggregateExec: mode=Partial, gby=[env@0 as env, region@1 as region, n@2 as n], aggr=[], lim=[30], ordering_mode=PartiallySorted([0])
        ProjectionExec: expr=[env@0 as env, region@1 as region, count(Int64(1)) PARTITION BY [t.env, t.deployment] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@3 as n]
          WindowAggExec: wdw=[count(Int64(1)) PARTITION BY [t.env, t.deployment] ...]
            SortExec: expr=[env@0 ASC NULLS LAST, deployment@2 ASC NULLS LAST], preserve_partitioning=[false]
              DataSourceExec: partitions=1, partition_sizes=[1]

The mechanism:

1. LimitedDistinctAggregation::transform_agg pushes LimitOptions::new(limit) into the AggregateExec. That value has no ordering direction (descending: None). The rule pushes it only while is_unordered_unfiltered_group_by_distinct() is true (source). The other producer of LimitOptions, TopKAggregation, uses LimitOptions::new_with_order and accepts only a single group key (source).
2. EnsureRequirements then inserts the SortExec that the window function needs below the aggregation. Because the child properties change, replace_children_if_necessary rebuilds the aggregation through AggregateExec::replace_children with ChildrenPropertiesMode::Recompute (source). This derives the properties again from the child, which is now sorted, but keeps limit_options. The rebuilt node gets ordering_mode=PartiallySorted([0]) and an output ordering, so is_unordered_unfiltered_group_by_distinct() is now false.
3. AggregateExec::execute_typed selects the top-k path when limit_options.is_some() && !is_unordered_unfiltered_group_by_distinct() (source). GroupedTopKAggregateStream::new needs a direction. It gets no direction from get_minmax_desc() (the query has no MIN/MAX) or from limit_options.descending, so it returns the internal error (source).

The top-k stream also reads only the first group key (aggr.group_expr().expr()[0]). Thus it would give wrong results for a multi-column DISTINCT even if a direction were available. The correct behavior is to not select that stream in this case.

I have a small fix in #25349.