Spark aggregate functions do not report distinct_handling

[mkleen]

Is your feature request related to a problem or challenge?

#25288 adds AggregateUDFImpl::distinct_handling (datafusion/expr/src/udaf.rs:958), which lets an aggregate declare whether DISTINCT can change its result, and EliminateAggregateDistinct uses that to drop the modifier from duplicate-insensitive functions. The default is DistinctHandling::Sensitive, which means the accumulator reads AccumulatorArgs::is_distinct and deduplicates its input, so leave the flag alone.

None of the aggregates in datafusion/spark overrides it, so they all fall back to Sensitive — and for three of them that is the wrong answer:

• collect_set (datafusion/spark/src/function/aggregate/collect.rs) always builds a DistinctArrayAggAccumulator, regardless of is_distinct. Deduplication is unconditional, so collect_set(DISTINCT x) and collect_set(x) return the same thing. That is the definition of DistinctHandling::Insensitive.
• collect_list (same file) always builds a plain ArrayAggAccumulator and never reads is_distinct. collect_list(DISTINCT x) therefore silently keeps duplicates instead of deduplicating — which per the enum docs is DistinctHandling::Unsupported, not Sensitive.
• try_sum (datafusion/spark/src/function/aggregate/try_sum.rs) likewise never reads is_distinct; TrySumAccumulator sums every value it is handed, so try_sum(DISTINCT x) silently returns the non-distinct sum. Also Unsupported.

The consequence today is a missed optimization for collect_set and, for collect_list / try_sum, wrong results rather than an error when a user writes DISTINCT.

Describe the solution you'd like

Override distinct_handling on the Spark aggregates to match what the accumulators actually do:

function	current (default)	correct
collect_set	Sensitive	Insensitive
collect_list	Sensitive	Unsupported
try_sum	Sensitive	Unsupported
avg	Sensitive	Unsupported

Then add coverage: a sqllogictest showing collect_set(DISTINCT x) planning as collect_set(x), and unit tests asserting each function's distinct_handling() so the tag and the accumulator cannot drift apart.

Describe alternatives you've considered

Give collect_list and try_sum real DISTINCT accumulators instead of tagging them Unsupported. That is strictly more work and does not have to block the tagging: nothing reads Unsupported yet, so tagging is a no-op at runtime today and becomes correct behavior for free once planning-time rejection lands. Tagging them now also records the gap in the code rather than in an issue.

[mkleen]

take