feat: integrate batch coalescer with async fn exec

[feniljain]

Which issue does this PR close?

• ClosesIntegrate BatchCoalescer into AsyncFuncExec and remove from CoalesceBatches optimization rule #19331

What changes are included in this PR?

Currently haven't included changes to remove it from optimizer, will do it in a subsequent PR as I see everyone else has done the same?

Mostly took code from other PRs and just stitched it together to work with AsyncFnExec

Are these changes tested?

In Progress

Are there any user-facing changes?

No

[feniljain]

Hey @jizezhang 👋🏻

I think we can review it now :)

[alamb]

Thanks @feniljain and @jizezhang -- I think this PR needs tests but otherwise could be merged in. I also have some ideas of how to simplify it a bit

[alamb]

Thank you @feniljain - this looks good to me

I am also interested in @jizezhang's comments too.

[alamb]

• Closes Integrate BatchCoalescer into AsyncFuncExec and remove from CoalesceBatches optimization rule #19331

BTW I think we'll need one more PR to remove the rule from CoalesceBatches (and maybe we can even remove the entire rule now...)

[jizezhang]

Looks good to me too, thank you.

[feniljain]

Thank you for patiently reviewing and helping me! 🙇🏻

BTW I think we'll need one more PR to remove the rule from CoalesceBatches (and maybe we can even remove the entire rule now...)

I will raise a PR for both of them after this gets merged. And also another PR to implement metrics in AsyncFuncExec :)

[alamb]

Thanks @feniljain and @jizezhang

[alamb]

Thank you for patiently reviewing and helping me! 🙇🏻

BTW I think we'll need one more PR to remove the rule from CoalesceBatches (and maybe we can even remove the entire rule now...)

I will raise a PR for both of them after this gets merged. And also another PR to implement metrics in AsyncFuncExec :)

Thank you @feniljain . I filed a ticket to track this:

• Remove CoalesceBatches optimizer rule #19591