feat: add multi level merge for sorting

[rluvaton]

Which issue does this PR close?

• Closes Reduce number of tokio blocking threads in SortExec spill #15323.

Rationale for this change

To be able to sort any amount spill files without getting over the tokio blocking thread limit

What changes are included in this PR?

Currently the PR is in draft as I want your opinions about the design.

1. Created a new Stream that use underneath the SortPreservingMergeStream
2. Change StreamingMergeBuilder to use the new stream when having spill files and spill manager

Are these changes tested?

Existing tests

Are there any user-facing changes?

Yes, kinda

[rluvaton]

Currently tokio don't expose the value of maximum number of blocking threads, and you have no way to know if you reached the limit as spawn_blocking doesn't return an indication that the current pool is full.

[rluvaton]

Need to add metrics in these cases

[rluvaton]

Need a better API than this, this is confusing as they are dependent

[rluvaton]

Will remove all the unwrap. first want to know what do you think about the solution

[2010YOUY01]

I think this is a similar problem as #14692, will check this out soon

[rluvaton]

I saw the pr from @ashdnazg and it is a better solution in my opinion as it does not need to know the number of available tokio tasks, and just make the read spill work

• Remove waits from blocking threads reading spill files. #15654

and although multi level merge sort should be implemented, @2010YOUY01 implementation (#15610) with adaption for row_hash in AggregateExec to use it might be the better way, so closing this

row_hash code to update:

datafusion/datafusion/physical-plan/src/aggregates/row_hash.rs

Lines 1053 to 1085 in 784df33

	fn update_merged_stream(&mut self) -> Result<()> {
	let Some(batch) = self.emit(EmitTo::All, true)? else {
	return Ok(());
	};
	// clear up memory for streaming_merge
	self.clear_all();
	self.update_memory_reservation()?;
	let mut streams: Vec<SendableRecordBatchStream> = vec![];
	let expr = self.spill_state.spill_expr.clone();
	let schema = batch.schema();
	streams.push(Box::pin(RecordBatchStreamAdapter::new(
	Arc::clone(&schema),
	futures::stream::once(futures::future::lazy(move |_| {
	sort_batch(&batch, expr.as_ref(), None)
	})),
	)));
	for spill in self.spill_state.spills.drain(..) {
	let stream = self.spill_state.spill_manager.read_spill_as_stream(spill)?;
	streams.push(stream);
	}
	self.spill_state.is_stream_merging = true;
	self.input = StreamingMergeBuilder::new()
	.with_streams(streams)
	.with_schema(schema)
	.with_expressions(self.spill_state.spill_expr.as_ref())
	.with_metrics(self.baseline_metrics.clone())
	.with_batch_size(self.batch_size)
	.with_reservation(self.reservation.new_empty())
	.build()?;
	self.input_done = false;
	self.group_ordering = GroupOrdering::Full(GroupOrderingFull::new());
	Ok(())
	}