PERF-#6433: Implement '.dropna()' using map-reduce pattern

[dchigarev]

This PR replaces a full-column implementation of .dropna() with a chain of two row-axis/map functions.

At the first stage, we submit a row-axis function evaluating whether each column should be dropped for this specific row partition and builds a boolean mask for that matter, at the second stage those masks are being merged and the columns to be dropped are being dropped.

perf results

Here's the comparison for df.dropna(axis=1, how="any") between these two implementations:

code to measure

import modin.pandas as pd
import modin.config as cfg
import numpy as np
from timeit import default_timer as timer

cfg.BenchmarkMode.put(True)

# start all the workers
pd.DataFrame([np.arange(cfg.MinPartitionSize.get()) for _ in range(cfg.NPartitions.get() ** 2)]).to_numpy()

shapes = [
    [5000, 5000],
    [1_000_000, 32],
    [10_000_000, 32],
    [10_000_000, 256]
]

for how in ("any",):
    for shape in shapes:
        # data-gen was copied from asv TimeDropna benchmark
        row, col = shape
        df = pd.DataFrame(np.random.randn(row, col))
        df.iloc[row // 20 : row // 10, col // 3 : col // 2] = np.nan
        df["foo"] = "bar"

        t1 = timer()
        df.dropna(axis=1, how=how)
        print(f"{shape=}, {how=}: {timer() - t1}")

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 Reimplement .dropna() using MapReduce pattern #6433
• tests are passing
• module layout described at docs/development/architecture.rst is up-to-date

[dchigarev]

Otherwise I was getting ~60s. for the case of a square-like frame with the shape of (5000, 5000), when normally it should pass for about 1-2 seconds

[dchigarev]

.copy_index_cache() doesn't trigger index materialization in comparison with ._get_index() that does it

[anmyachev]

Let's not mix abstraction levels and separate this condition into a dataframe method.

[dchigarev]

Let's separate this condition into a dataframe method.

well, that would be quite hard...

here I want to dispatch between two different implementations of .dropna(). Since dataframe level doesn't define dropna, then I have to do this dispatching at the QC level.

The idea of a "general dispatcher" at the dataframe level was discussed at #5394 but trying suggested ideas didn't really find success, so there's a lot of work to be done in this regard

[anmyachev]

just hide implementation details something like that:

        processable_amount_of_partitions = self._modin_frame.processable_amount_of_partitions_for_dropna

[dchigarev]

Except for the .processable_amount_of_partitions_for_dropna field, I simply added a property .num_parts and kept the deciding logic at the QC, so the line now looks like:

processable_amount_of_partitions = self._modin_frame.num_parts < CpuCount.get() * 32

It feels wrong adding the .processable_amount_of_partitions_for_dropna field, as the modin_frame layer should know nothing about the dropna and its implementation desires in terms of partitions amount

[anmyachev]

LGTM!