FEAT-#7118: Add range-partitioning impl for 'df.resample()'

[dchigarev]

What do these changes do?

Adds range-partitioning impl for df.resample(). The new implementation doesn't always work better, so enabling it only when the flag is specified.

script to measure

import pandas
import numpy as np
import modin.pandas as pd
import modin.config as cfg

from timeit import default_timer as timer

from modin.utils import execute

cfg.CpuCount.put(16)

nrows = [1_000_000, 5_000_000, 10_000_000]
ncols = [5, 33]
rules = [
    "500ms", # doubles nrows
    "30s", # decreases nrows in 30 times
    "5min", # decreases nrows in 300
]
use_rparts = [True, False]

cols = pandas.MultiIndex.from_product([rules, ncols, use_rparts], names=["rule", "ncols", "USE RANGE PART"])
rres = pandas.DataFrame(index=nrows, columns=cols)

total_nits = len(nrows) * len(ncols) * len(rules) * len(use_rparts)
i = 0

for nrow in nrows:
    for ncol in ncols:
        index = pandas.date_range("31/12/2000", periods=nrow, freq="s")
        data = {f"col{i}": np.arange(nrow) for i in range(ncol)}
        pd_df = pandas.DataFrame(data, index=index)
        for rule in rules:
            for rparts in use_rparts:
                print(f"{round((i / total_nits) * 100, 2)}%")
                i += 1
                cfg.RangePartitioning.put(rparts)

                df = pd.DataFrame(data, index=index)
                execute(df)

                t1 = timer()
                res = df.resample(rule).sum()
                execute(res)
                ts = timer() - t1
                print(nrow, ncol, rule, rparts, ts)

                rres.loc[nrow, (rule, ncol, rparts)] = ts
                rres.to_excel("resample.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 df.resample() #7118
• tests added and passing
• module layout described at docs/development/architecture.rst is up-to-date

[YarShev]

Why is this import not at the top of the file?

[dchigarev]

The answer is right above in the codeQL warning :)

it triggers a circular import otherwise

[dchigarev]

an example of why it's requires

Imagine we have a time series with an Hour resolution:

>>> sh
                       a
2018-01-01 00:00:00  0.0
2018-01-01 01:00:00  1.0
2018-01-01 02:00:00  2.0
2018-01-01 03:00:00  3.0

Resampling this into 30-min intervals gives this:

>>> expected_res = sh.resample("30min").sum()
>>> expected_res
                       a
2018-01-01 00:00:00  0.0
2018-01-01 00:30:00  0.0  <---- interpolated value
2018-01-01 01:00:00  1.0
2018-01-01 01:30:00  0.0 <---- interpolated value
2018-01-01 02:00:00  2.0
2018-01-01 02:30:00  0.0 <---- interpolated value
2018-01-01 03:00:00  3.0

Let's now emulate parallel execution of resample and split sh into two partitions:

>>> pd.concat([sh.iloc[:2].resample("30min").sum(), sh.iloc[2:].resample("30min").sum()])
                       a
2018-01-01 00:00:00  0.0
2018-01-01 00:30:00  0.0  <---- interpolated value
2018-01-01 01:00:00  1.0
*should be an interpolated value here, but it's missing*
2018-01-01 02:00:00  2.0
2018-01-01 02:30:00  0.0 <---- interpolated value
2018-01-01 03:00:00  3.0

The reason for the missing value is that, the first partition only sees an interval from [00:00:00, 01:00:00] and the second sees [02:00:00, 03:00:00], so it's unclear that we should fill the gap between 1h and 2h with 1:30.

One of the solutions is to insert a timestamp dummy value with a slight offset in each partition, so it would know the real bounds of the partition.

[dchigarev]

more data to actually use partitioning

[dchigarev]

was missing before, however is needed to function properly

[dchigarev]

this approach doesn't work well with transform operations, so all of them are disabled

[YarShev]

Suggested change

	closed_on_right : bool, default: False
	close_to_right : bool, default: False

?

[dchigarev]

here I refer to the term "closed interval", closed_on_right means, that we have to include the right bound in it

[YarShev]

Suggested change

	allow_range_impl : bool, default: True
	range_impl : bool, default: True

or use_range_impl?

[dchigarev]

allow_range_impl=True doesn't necessarily mean, that the range-partition will be used, it also depends on cfg.RangePartitioning value and axis argument. So this parameter indeed only 'allows' for the range-partitioning to be used, not dictates that.