fix: keep aggregate limits when combining or rebuilding AggregateExec

[jayzhan211]

Which issue does this PR close?

• No separate issue. Both problems are regressions from refactor: Make AggregateExec state modeling and public updates safe (part2: topk) #25696.

Rationale for this change

1. Some GROUP BY ... LIMIT queries fail to plan. With one target partition:

   SET datafusion.execution.target_partitions = 1;
   CREATE TABLE t AS SELECT * FROM (VALUES (1, 10), (2, 20), (3, 30), (4, 40), (5, 50), (6, 60)) AS v(a, b);
   SELECT b, a FROM t GROUP BY b, a LIMIT 2;
   SELECT a % 3 FROM t GROUP BY a % 3 LIMIT 2;

   both fail with:

   DataFusion error: CombinePartialFinalAggregate
   caused by
   Internal error: The AggregateKind should stay either (a) General (b) DistinctLimit introduced with the previous optimizer pass `LimitedDistinctAggregation`, it's impossible to have other variant.
   

   When the final aggregate's grouping differs from the partial's (expression keys, or columns not at their output positions), LimitedDistinctAggregation limits only the final aggregate. CombinePartialFinalAggregate treats that pair as impossible and returns an error.

2. On Parquet tables the TopK aggregate limit is dropped. EXPLAIN SELECT DISTINCT k FROM pt ORDER BY k LIMIT 10 keeps lim=[10] on the aggregates for a memory table but not for a Parquet table, so the query falls back to a full hash aggregation. MIN/MAX TopK aggregates (ORDER BY max(x) DESC NULLS LAST LIMIT n) lose their limit the same way. Post-optimization filter pushdown rebuilds the Parquet scan when it accepts a dynamic filter, and rebuilding the aggregates above it re-applies only the DISTINCT limit, not the TopK ones.

Before #25696 both places carried the limit over with with_limit_options(...limit_options()).

What changes are included in this PR?

• CombinePartialFinalAggregate: if the final aggregate has a DISTINCT limit, re-apply it to the combined aggregate with try_optimize_distinct_soft_limit, which checks eligibility again. Otherwise keep the combined aggregate as built.
• AggregateExec::replace_children (ChildrenPropertiesMode::Recompute): also re-apply TopKMinMax and TopKDistinct through try_optimize_topk, using the stored limit, direction and NULL placement.

What is the testing strategy for this PR?

New sqllogictest cases. Each one fails without the fix and passes with it:

• aggregate.slt: both queries above at target_partitions = 1. An EXPLAIN checks the combined mode=Single ... lim=[2] aggregate, and a row-count check keeps the results deterministic.
• aggregates_topk.slt: a Parquet-backed table. EXPLAIN checks that lim=[10] stays on both aggregates for SELECT DISTINCT k ... ORDER BY k LIMIT 10, and that lim=[2] stays for a MAX TopK query whose scan gets a semi-join dynamic filter. Result checks are included.

Are there any user-facing changes?

No API changes. The listed queries plan again, and TopK aggregation applies again on Parquet scans that receive dynamic filters.

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 82.64%. Comparing base (801b017) to head (290c3fc).
⚠️ Report is 7 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #25983      +/-   ##
==========================================
+ Coverage   82.62%   82.64%   +0.01%     
==========================================
  Files        1147     1147              
  Lines      445320   445520     +200     
  Branches   445320   445520     +200     
==========================================
+ Hits       367964   368184     +220     
+ Misses      55013    54975      -38     
- Partials    22343    22361      +18     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[2010YOUY01]

Thank you for the fix!

For two issues:

1. It seem like a deeper bug in another optimizer rule, and I suggested a simpler fix in comment
2. LGTM

So I suggest to use the simpler fix here for issue 1, and this PR should be good to go, and we can fix the remaining problem in a follow-up PR.

[2010YOUY01]

This fix looks like patching another bug in previous optimizer rule LimitedDistinctAggregation

Its contract was

# Before
Limit(k)
-- AggregateExec(final)
---- AggregateExec(partial)

# After
Limit(k)
-- AggregateExec(final, limit=k)
---- AggregateExec(partial, limit=k)

And this fix seem to accept the mixed case, I'm not sure if there is some workload can get optimized to this shape, and also combining it to a AggregateExec(single, limit=k) is safe

# After
Limit(k)
-- AggregateExec(final, limit=k)      <-- with limit
---- AggregateExec(partial)             <-- no limit

[2010YOUY01]

The alternative fix for the internal error is to let this branch return not transformed, though it misses some optimization oppurtunity.

[jayzhan211]

Comments addressed!

[2010YOUY01]

thanks again

[2010YOUY01]

1. It seem like a deeper bug in another optimizer rule, and I suggested a simpler fix in comment

I got a fix for it, I will open a PR after this PR has merged to reuse tests.

I'll merge this PR tomorrow.