[Epic] Optionally Limit memory used by DataFusion plan

[alamb]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.

The basic challenge is that as of today, DataFusion can use an unbounded amount of memory for running a plan and it is neither possible to calculate the memory before hand nor limit the use.

If DataFusion processes individual partitions that are larger than the available memory system memory, right now it will keep allocating memory from the system until it is killed by the OS or container system.

Also, when running multiple datafusion plans in the same process, each will consume memory without limit where it may be desirable to reserve / cap memory usage by any individual plan to ensure the plans don't together exceed the system memory budget

Thus, it would be nice if we could give DataFusion's plans a memory budget which they then stayed under

Describe the solution you'd like

1. Add an option to ExecutionConfig that has a “total plan memory budget”
2. Add logic to each node that requires a memory buffer to ensure it stays under the limit.

The operators that can use large amounts of memory today are:

1. Sort
2. Join
3. GroupByHash

There are many potential ways to ensure the limit is respected:

1. (Simplest) error if the budget is exceeded
2. (more complex): employ algorithms that can use secondary storage (e.g. temp files) like sort that spills multiple round of partial sorted results and give a final merge phase for the partition global ordering

Describe alternatives you've considered
There are some interesting tradeoffs between “up front allocation” dividing memory up across all operators that would need it and a more dynamic approach.

This is likely something that will require some major efforts over many different issues -- I suggest we use this issue to implement a simple "error if over limit" strategy and then work on more sophisticated strategies subsequently

Progress tracking

Added Jan 2022:

Remaining Work

• Initial MemoryManager and DiskManager APIs for query execution + External Sort implementation #1526
• [Epic] Generate runtime errors if the memory budget is exceeded #3941
• [EPIC] Memory Limited Sort (Externalized / Spill) #1568
• Memory Limited GroupBy (Externalized / Spill) #1570
• Memory Limited Joins (Externalized / Spill) #1599
• Track memory usage in Non Limited Operators #1569
• DiskManager Performs Blocking IO #1637
• Unlimited memory consumption in RepartitionExec #4816

[edrevo]

I would add Repartition as another operation that might use a bunch of memory.

[andygrove]

We should also discuss creating a scheduler in DataFusion (see #64) since it is related to this work. Rather than try and run all the things at once, it would be better to schedule work based on the available resources (cores / memory). We would still need the ability to track/limit memory use within operators but the scheduler could be aware of this and only allocate tasks if there is memory budget available.

[alamb]

I filed #898 for tracking memory used by a plan

[alamb]

#899 for tracking memory used by individual operators

[yjshen]

I created a proposal trying to fix this. Please refer to https://docs.google.com/document/d/1BT5HH-2sKq-Jxo51PNE6l9NNd_F-FyyYcyC3SKTnkIA/edit# for the whole proposal.

[alamb]

I have started added a "Progress Tracking" list to the description of this ticket. Please update it with additional items as you discover them.