FEAT-#7047: Add range-partitioning implementation for '.pivot_table()'

[dchigarev]

What do these changes do?

This PR adds a range-partitioning implementation for .pivot_table() method. Pivot table is literally a groupby aggregation + fancy post-processing of the result.

The new implementation uses range-partitioning groupby to perform at the first stage and then applies make_pivot_table() to the reduced result.

Range-partitioning implementation seems to outperform the old full-column implementation on a normal-size data. That's why I decided to replace the old full-column impl with range-partitioning everywhere where possible:

script to measure

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

import modin.config as cfg
cfg.CpuCount.put(44)
from modin.utils import execute

nrows = [100_000, 1_000_000, 2_500_000, 5_000_000, 10_000_000]
ncols = 34
values = ["value0", [f"value{i}" for i in range(5)]]
ngroups = [10, 1_000]
impl = ["full_axis", "map_reduce", "range_part"]

def get_num_vals(val):
    return (
        len(val)
        if isinstance(val, list)
        else (1 if isinstance(values, str) else ncols - 3)
    )

columns = pandas.MultiIndex.from_product(
    [
        [
            get_num_vals(val)
            for val in values
        ],
        ngroups,
        impl,
    ],
    names=["num_values", "num_groups", "impl"],
)
total_res = pandas.DataFrame(index=nrows, columns=columns)

i = 0
total_its = len(nrows) * len(values) * len(ngroups) * len(impl)

for nrow in nrows:
    for val in values:
        for ngroup in ngroups:
            data = {
                "index1": np.tile(np.arange(ngroup), nrow // ngroup),
                "index2": np.tile(np.arange(ngroup), nrow // ngroup),
                "col1": np.tile([f"val{i}" for i in range(ngroup)], nrow // ngroup),
                **{f"value{i}": np.arange(nrow) for i in range(ncols - 3)},
            }

            for imp in impl:
                print(f"{round((i / total_its) * 100, 2)}%")
                i = i + 1
                df = pd.DataFrame(data)
                execute(df)

                t1 = timer()
                res = df.pivot_table(
                    index=["index1", "index2"],
                    columns=["col1"],
                    values=val,
                    # requires a hack in 'pivot_table()' that would dispatch to a proper implementation
                    # depending on this parameter
                    margins_name=imp,
                )
                execute(res)
                tm = timer() - t1
                print(f"{nrow=}; {val=}; {ngroup=}; {imp=}: {tm}; {res.shape}")
                total_res.loc[nrow, (get_num_vals(val), ngroup, imp)] = tm
                total_res.to_excel("pivot.xlsx")

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

[dchigarev]

.pivot_table() is literally a groupby + fancy post-processing, so decided to put it into groupby.py

[dchigarev]

the logic was copied from qc.pivot_table()

[dchigarev]

the logic was copied from PandasQueryCompiler._pivot_table_tree_reduce()

[dchigarev]

the logic was copied from PandasQueryCompiler.pivot_table

[anmyachev]

This PR adds a range-partitioning implementation for .pivot_table() method and enables it by default.

@dchigarev this confuses me a little, because as far as I understand MapReduce implementation is by default, right?

[dchigarev]

This PR adds a range-partitioning implementation for .pivot_table() method and enables it by default.

@dchigarev this confuses me a little, because as far as I understand MapReduce implementation is by default, right?

Right, the order is the following:

1. Try MapReduce implementation
2. If can't use MapReduce, use range-partitioning impl
3. If can't use range-partitioning, use full-column impl

Agree that the comment is a bit confusing, rephrased it

[anmyachev]

Could you also rebase on master? To be sure that the new tests are passed.

[anmyachev]

LGTM!