Spill error (oom) with sort merge join

[xxhZs]

Describe the bug

I run some data(300g) in small mem(64g) with sort merge join spill.
Then i get error
Failed to allocate additional {} for {} with {} already allocated for this reservation - {} remain available for the total pool

And i try to fix it. I fix two code
remove this code:

datafusion/datafusion/physical-plan/src/sorts/stream.rs

Line 176 in c89cb70

self.reservation.try_resize(self.converter.size())?;

and

datafusion/datafusion/physical-plan/src/sorts/sort.rs

Line 785 in c89cb70

match self.reservation.try_grow(size) {

to

loop {
match self.reservation.try_grow(size) {
Ok(_) => Ok(()),
Err(e) => {
if self.in_mem_batches.is_empty() {
return Err(Self::err_with_oom_context(e));
}

            // Spill and try again.
            self.sort_and_spill_in_mem_batches().await?;
    }

}

I think the issues are as follows:

1. The memory was immediately occupied as soon as the sort operator spilled data to disk.
2. The row operator failed to spill data to disk, which caused this problem.
   So could we fix it in this way, or are there any better solutions?

To Reproduce

No response

Expected behavior

No response

Additional context

No response

[xudong963]

Would it be easy to have a reproduced case? Which is helpful to add as a test in your fix

[xxhZs]

Would it be easy to have a reproduced case? Which is helpful to add as a test in your fix

could you tell me how to write some test data in DataFusion? I'd like to insert some data to run tests on my side.