Reduce number of tokio blocking threads in SortExec spill

[andygrove]

Is your feature request related to a problem or challenge?

In Comet, we see some queries "hang" when running with minimal memory. The issue appears to be that we have hundreds of spill files and each spill requires its own tokio blocking thread and Comet does not have enough threads available.

See apache/datafusion-comet#1523 (comment) for more detail.

Describe the solution you'd like

No response

Describe alternatives you've considered

No response

Additional context

No response

[alamb]

Do you see too many threads when writing the spill files or when reading?

[andygrove]

Do you see too many threads when writing the spill files or when reading?

This is when reading, during the merge operation.

In merge phase, each spill file will be wrapped by a stream backed by a blocking thread (see read_spill_as_stream), so we'll spawn at least 183 blocking threads when there are 183 spill files to merge spilled data.

[alamb]

Makes sense -- with 183 spill files, we probably would need to merge in stages

For example starting with 183 spill files

1. run 10 jobs, each merging about 10 files into one (results in 10 files)
2. run the final merge of 10 files

This results in 2x the IO (have to read / write each row twice) but it would be possible at least to parallelize the merges of the earlier step

I think @2010YOUY01 was starting to look into a SpillFileManager -- this is the kind of behavior I would imagine being part of such a thing

[rluvaton]

I think I have the the same problem but in AggregateExec when using row_hash, as it spills as well and use SortPreservingMergeStream.

I think the solution should actually be in SortPreservingMergeStream rather than SpillFileManager no? although it does not spawn blocking threads it should support the multiple levels to merge

[alamb]

I think I have the the same problem but in AggregateExec when using row_hash, as it spills as well and use SortPreservingMergeStream.

I think the solution should actually be in SortPreservingMergeStream rather than SpillFileManager no? although it does not spawn blocking threads it should support the multiple levels to merge

I am not sure / familiar enough with the code to know off the top of my head.

I do think having hash and sort use the same codepath (that we can then go optimize a lot) sounds like a great idea

[rluvaton]

I have a working version locally and will create a PR soon.

there is a problem though, tokio don't expose the maximum number of blocking threads, and if you try to call spawn_blocking while there are no available threads, no error will be returned.

this is important as for example Comet set this by default to 10, and tokio default is 512 IIRC.

the working version can be improved with some optimization like prefetch and more, but it will be good enough for now and we can iterate further

[andygrove]

I have a working version locally and will create a PR soon, just one problem, I don't think we can know the number of blocking threads tokio is configured with.

this is important as for example Comet set this by default to 10, and tokio default is 512 IIRC.

the working version can be improved with some optimization like prefetch and more, but it will be good enough for now and we can iterate further

Comet currently creates a new tokio runtime per plan but there is a proposal to move to a global tokio runtime (per executor) instead.

apache/datafusion-comet#1590

[rluvaton]

Comet currently creates a new tokio runtime per plan but there is a proposal to move to a global tokio runtime (per executor) instead.

apache/datafusion-comet#1590

even if you use global tokio runtime and set the number of blocking threads to be a 1000 for example, there can be 1001 spill files. the problem is the same

[alamb]

even if you use global tokio runtime and set the number of blocking threads to be a 1000 for example, there can be 1001 spill files. the problem is the same

At some point the system is going to be IO bound so having more blocking threads doing I/O isn't going to help IO and will likely consume non trivial time context switching between them

I think a better solution is to more carefully manage how many files are being spilled / read at any time. This will be more complicated (as we'll likely have to do multiple merge phases, etc) but I think it is a better approach in the long run