FEAT-#6965: Implement '.merge()' using range-partitioning implementation

[dchigarev]

What do these changes do?

This PR adds a new implementation for merge() using range-partitioning mechanism. The new implementation does the following:

1. Computes range-partition bins for left dataframe
2. Builds range-partitioning for left dataframe using bins computed at the first step
3. Builds range-partitioning for right dataframe using bins computed at the first step
4. Applies the merge kernel row-wise to the left dataframe and broadcasts row partitions of the right dataframe there

The benefit of this implementation is that we never gather the whole right dataframe in one partition (as we do it now) but rather repartition it in a way so it would be correct to broadcast only row partitions of the right df. This implementation benefits when the right dataframe is relatively big.

One of the downsides of this implementation is that repartitioning changes the order of rows, meaning that the result is always sorted by keys at the end.

At the moment, one can only use range-partitioning impl for merge by specifying a special config variable (cfg.RangePartitioningMerge).

Perf measurements for h2o
Important note: original h2o dataset has several categorical columns with high cardinality (high amount of unique values), this is a problematic case for modin (see 2nd point) and it so it works terribly slow with such dtypes. In the measurements below, all categorical columns were casted to strings.

h2o joins, 500mb data

h2o joins, 5gb data

• 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 merge() using range-partitioning implementation #6965
• tests added and passing
• module layout described at docs/development/architecture.rst is up-to-date

[dchigarev]

this function was copied without changes from PandasQueryCompiler.merge

[dchigarev]

this logic was copied without any changes from PandasQueryCompiler.merge and was placed in a separate method to be reused by range-partitioning impl

[dchigarev]

We're planning to implement more methods using range-partitioning and give users a choice to switch between implementations on their own. The approach with config variables doesn't seem to scale good enough, as creating a config variable for each method isn't a good idea IMO.

An alternative could be passing some parameter at the pandas API level specifying which implementation to use:

df1.merge(df2, on="key", impl="range-partitioning")
df.groupby(...).apply(..., impl="range-partitioning")
df.nunique(impl="range-partitioning")

The problems with this approach are:

1. The user's code loses compatibility with pandas (switching back from modin to pandas would require removing those extra arguments)
2. The parameter will only make sense for executions based on PandasQueryCompiler (snowflake, hdk executions won't be able to support it)

What do others think in this regard?

[anmyachev]

What do others think in this regard?

Why not just use one config variable (like RangePartitioningImpl) for all the functions that support it?

[dchigarev]

Why not just use one config variable (like RangePartitioningImpl) for all the functions that support it?

Yep, it seems like the only option. We could have then expanded this page, giving a short description of every operation that supports range-partitioning with tips on when to use it, and then probably link this page in our optimization notes . I'm personally voting for this option, will try to implement it in this PR. cc @YarShev

[dchigarev]

Added a common variable RangePartitioning, let's handle the transition from RangePartitioningGroupby as well as changes in documentation in a separate PR

[anmyachev]

Why?

[dchigarev]

added the following in-code comment:

# pandas resets the index of the result unless we were merging on an index level,
# the current implementation only supports merging on column names, so dropping
# the index unconditionally

[anmyachev]

LGTM!