SQL: lower DISTINCT ON with ordered first-row-per-group semantics #289

Context

Split from #111. Every other item in that mixed SQL-gap issue is resolved or deliberately rejected:

• median / approx_median lower to quantile;
• array_agg remains deliberately unsupported;
• supported IN / EXISTS forms landed, with remaining subquery semantics tracked by SQL: NOT IN, correlated IN, and scalar subqueries remain unsupported #125.

The only unresolved item is PostgreSQL-style DISTINCT ON:

SELECT DISTINCT ON (service)
       service, ts, latency
FROM metrics
ORDER BY service, ts DESC;

DISTINCT ON (keys) keeps the first row in each key group according to the query ordering. It is not equivalent to ordinary full-row DISTINCT, and its result is nondeterministic when the ordering does not fully select a row among peers.

Required semantics

• Preserve the DISTINCT ON key expressions separately from ordinary DISTINCT columns.
• Preserve the ordering that determines the selected row.
• Define behavior when no ORDER BY is present or when its leading expressions do not match the DISTINCT ON expressions, matching the selected SQL dialect/DataFusion contract.
• Do not lower it to a plain Dedup, because Dedup does not encode which ordered row survives.

Possible canonical representation

Prefer expressing it through existing relational operators if semantics remain lossless, for example a per-partition row-number/top-1 shape:

Window(row_number, partition_by = distinct_on_keys, order_by = query_order)
  -> Filter(row_number = 1)
  -> Project(original output)

If the existing canonical TopK/Limit representation can express this exactly, reuse that representation rather than adding a PostgreSQL-specific node. The chosen representation must retain deterministic tie/order semantics.

Scope

• Lower DataFusion/PostgreSQL DISTINCT ON into lossless canonical pre-ASAP IR.
• Preserve schema, nullability, ordering, and key identity.
• Thread the shape through resolution, CSE, DAG export, and existing canonicalization.
• Add positive and negative SQL lowering tests.

Acceptance criteria

• A representative DISTINCT ON (service) ... ORDER BY service, ts DESC query lowers successfully.
• Multiple keys and qualified columns lower correctly.
• Ordinary SELECT DISTINCT behavior is unchanged.
• The chosen row follows ordering semantics and is not modeled as unordered deduplication.
• Invalid or unsupported ordering/dialect combinations return an explicit diagnostic.
• Serialization/CSE and DAG export preserve the partition keys and ordering.

Related

• Split from SQL: subquery-valued predicates (IN/EXISTS/scalar), DISTINCT ON, and median/array_agg unsupported #111.
• Remaining subquery-valued SQL expressions are SQL: NOT IN, correlated IN, and scalar subqueries remain unsupported #125.
• Broader corpus coverage is Add corpus tests from DQC, BGP analytics, ClickBench, and business analytics workloads #132.