FEAT-#6929: Implement Series.case_when in a distributed way

[AndreyPavlenko]

What do these changes do?

• first commit message and PR title follow format outlined here

  NOTE: If you edit the PR title to match this format, you need to add another commit (even if it's empty) or amend your last commit for the CI job that checks the PR title to pick up the new PR title.

• passes flake8 modin/ asv_bench/benchmarks scripts/doc_checker.py
• passes black --check modin/ asv_bench/benchmarks scripts/doc_checker.py
• signed commit with git commit -s
• Resolves Implement Series.case_when in a distributed way #6929
• tests added and passing
• module layout described at docs/development/architecture.rst is up-to-date

[AndreyPavlenko]

I still don't like the idea of a method being responsible to cache itself. We should either have a common mechanism that does this automatically or not to have this logic at all. Does the caching gives any significant improvement? Can we remove it?

I've not benchmarked, but it should be more efficient, because, if not cached, the same function is serialized multiple times for each partition on each call.

[dchigarev]

but it should be more efficient

There's no doubt that it's more efficient, but my point is that, in reality, nobody runs .case_when() thousand times in a row. Even if one does, they would get ~100-200ms saving because of the caching. This may worth it if implemented properly (like automatical caching in RayWraper._funcs_cache), but I personally can't justify the current approach with a method manually caching itself in a dataframe's attribute.

Seems, for Dask it does not work. If the function is cached, the tests randomly fail with CancelledError.

This should actually be considered as a huge red flag, pointing out that there could be other unexpected side-effects of such implementation of caching with other engines.

if not cached, the same function is serialized multiple times for each partition

Let's call PartitionMaganer.preprocess_func() in the beginning of PandasDataframe.case_when(), this is a common way of how other dataframe methods 'cache' a kernel so each partition would reuse it.

[YarShev]

@AndreyPavlenko, what else is required to be done in this PR? Please also resolve conflicts.

[YarShev]

@anmyachev, @dchigarev, any comments?