Integrate `BatchCoalescer` into `AsyncFuncExec` and remove from `CoalesceBatches` optimization rule

[alamb]

Is your feature request related to a problem or challenge?

• Part of [EPIC] Remove CoalesceBatchesExec operator #18779

The LimitedBatchCoalescer can now be used to integrate batch coalescing directly in ExecutionPlan implementations, which is better than using an external operator for the reasons described on #18779

Currently AsyncFuncExec needs a CoalesceBatches operator after for performance, see

datafusion/datafusion/physical-optimizer/src/coalesce_batches.rs

Line 78 in bd30fe2

} else if let Some(async_exec) = plan_any.downcast_ref::<AsyncFuncExec>() {

We should integrate coalescing directly in the operator

Describe the solution you'd like

1. Add batch coalescing into AsyncFuncExec
2. Remove AsyncFuncExec from the CoalesceBatches optimizer rule:

   datafusion/datafusion/physical-optimizer/src/coalesce_batches.rs

   Line 78 in bd30fe2

   } else if let Some(async_exec) = plan_any.downcast_ref::<AsyncFuncExec>() {

Describe alternatives you've considered

@jizezhang did this for RepartitionExec here:

• remove repartition exec from coalesce batches optimizer #19239

@Dandandan did this for FilterExec in these two PRs, so we can probably follow a similar model

• Coalesce batches inside FilterExec #18604
• Remove FilterExec from CoalesceBatches optimization rule, add fetch support #18630

Additional context

No response

[feniljain]

take

[alamb]

BTW I think @@jizezhang may also be working on this

[jizezhang]

I was about to, but I saw @feniljain took it. Please feel free to continue @feniljain . and thanks for mentioning @alamb :)

[alamb]

I was about to, but I saw @feniljain took it. Please feel free to continue @feniljain . and thanks for mentioning @alamb :)

Perhaps you would be willing to help review the PR @jizezhang ? It looks like @feniljain has started here

• wip: integrate batch coalescer with async fn exec #19342