feat: track memory usage for recursive CTE, enable recursive CTEs by default

[jonahgao]

Which issue does this PR close?

Closes #9554.

Rationale for this change

1. Add a MemoryReservation to track the memory usage of the RecursiveQuery's buffer.
2. When a round of iteration ends, the buffer and its MemoryReservation are moved together to the WorkTable, and then passed on to the MemoryStream in the next iteration.
3. The MemoryReservation will finally be freed after the MemoryStream is dropped.

What changes are included in this PR?

• Track memory usage for recursive CTEs.
• Enable recursive CTE support by default.

Are these changes tested?

Yes

Are there any user-facing changes?

No. Enable a new feature by default.

[alamb]

Very nice -- thank you @jonahgao

cc @matthewgapp

[alamb]

👍

[alamb]

I was a little worried at first that this optional API makes it easy to forget to provide reservation. However I see now that the reservation is used only with the recursive CTE case

(not for this PR) In general I wondered if we should always have a memory reservation to MemoryStream 🤔 I think that would double count batches from a MemTable however, so it isn't an obviously good improvement

[jonahgao]

This concern is quite reasonable; previously there was no need to add a MemoryReservation to the MemoryStream, and the with_reservation function is a bit confused.

I plan to no longer use MemoryStream for future recursive CTEs and instead use a new stream class that can read the WorkTable multiple times, in order to support
https://github.com/apache/arrow-datafusion/blob/81b0a011705b17a09f494f550a5382b0c3414597/datafusion/physical-plan/src/recursive_query.rs#L316

So that the MemoryStream::with_reservation API can be removed.

[matthewgapp]

Looks great, @jonahgao—thanks for taking this on. Would it be sensible to have the worktable's update method take a ReservedBatches as an argument, instead of constructing it internally? My thinking is that, given worktable's sole purpose is to facilitate the recursive query, tightening the integration with ReservedBatches could better illustrate to future contributors the coupled ownership relationship between a RecordBatch's buffer's and its corresponding memory reservation. It's more of a nit than anything so feel free to ignore :)

And apologies that I've been a bit radio silent on this front @alamb! Been heads down on other stuff :/

[alamb]

And apologies that I've been a bit radio silent on this front @alamb! Been heads down on other stuff :/

No worries @matthewgapp -- I totally understand. Good luck and thank you for contributing the original PR 🙏

[jonahgao]

Looks great, @jonahgao—thanks for taking this on. Would it be sensible to have the worktable's update method take a ReservedBatches as an argument, instead of constructing it internally? My thinking is that, given worktable's sole purpose is to facilitate the recursive query, tightening the integration with ReservedBatches could better illustrate to future contributors the coupled ownership relationship between a RecordBatch's buffer's and its corresponding memory reservation. It's more of a nit than anything so feel free to ignore :)

And apologies that I've been a bit radio silent on this front @alamb! Been heads down on other stuff :/

Make sense to me! Updated and thank you @matthewgapp

[alamb]

Thanks @jonahgao and @matthewgapp 🙏