Resolve lost wakeup in SpillPoolReader with multiple concurrent SpillPoolWriters

[pepijnve]

Which issue does this PR close?

• Closes SMJ + aggregate finish with one partition but stuck forever when number of partitions is big #23447

Rationale for this change

When multiple SpillPoolWriter clones concurrently push batches to the same channel, more than one non-finished SpillFile can be in flight. This happens because each SpillPoolWriter clone takes the current_write_file at the start of push_batch and puts it back when it's done. When multiple push_batch calls happen concurrently, only the first one will be able to take the current_write_file and the others will all create their own new spill file. Which one gets put back for subsequent use is a race condition.

If this occurred and the writers are all dropped before rotation happens in, multiple files in the files deque will be have writer_finished == false. The last writer drop logic in SpillPoolWriter::drop only finishes whatever file is the current_write_file as finished.

This can lead to a stalled situation when SpillPoolFile::poll_next catches up with the writer and returns Pending because writer_finished == false. A waker for the file is registered, but since the last writer drop logic only finishes and wakes whatever happens to be current_write_file, which may not be the current read file, the waker may end up never being notified.

There is a secondary waker that is registered on the spill pool itself, but due to fine grained locking, it is possible for the wake call in the last writer drop logic to be called before the waker registration.

What changes are included in this PR?

• Add support for tracking multiple unfinished write files
• Close all unfinished write files when the last writer is dropped
• Removed writer_dropped field which was an unnecessary denormalisation of active_writer_count == 0

An additional benefit of tracking all unfinished write files is that excessive creation of tiny spill files is avoided when many writers are pushing batches concurrently.

Are these changes tested?

Reproduction case from linked issue was used to confirm fix

Are there any user-facing changes?

No

[SemyonSinchenko]

An additional benefit of tracking all unfinished write files is that excessive creation of tiny spill files is avoided when many writers are pushing batches concurrently.

That is interesting because I constantly facing the "Too many open files" when there is a heavy spill. I did not know it is a problem and I was thinking it is OK, just used ulimit -n 524288.