Repository: modin-project/modin
Pull request: PERF-#6876: Skip the masking stage on 'iloc' where beneficial
<!--
Thank you for your contribution!
Please review the contributing docs: https://modin.readthedocs.io/en/latest/development/contributing.html
if you have questions about contributing.
-->

## What do these changes do?

This PR extends the idea originally introduced in #6423.

<b> What's the idea </b>

The masking pipeline consists of two stages:
1. Compute individual indexers for each partition and mask them.
2. If the indexing forces reordering, apply a full-column reordering function to the result of the first stage.

In theory, the second stage can function without the first one, the first stage is needed in order to reduce the dimension of the input for the second stage. However, sometimes the first stage can be quite expensive, and it could be reasonable to skip it and jump to the second one right away.

This PR extends the number of cases where we skip the first stage with the cases where we know for sure that the reordering stage will be required. Based on the measurements below, I found out the following condition that controls this:

```python
# 'base_num_cols' specifies the number of columns that the dataframe should have
# in order to jump to 'reordered_labels' in case of len(row_positions) / len(self) >= base_ratio;
# these variables may be a subject to change in order to tune performance more accurately
base_num_cols = 10
base_ratio = 0.2
# Example:
#   len(self.columns): 10 == base_num_cols -> min ratio to jump to reorder_labels: 0.2 == base_ratio
#   len(self.columns): 15 -> min ratio to jump to reorder_labels: 0.3
#   len(self.columns): 20 -> min ratio to jump to reorder_labels: 0.4
#   ...
#   len(self.columns): 49 -> min ratio to jump to reorder_labels: 0.98
#   len(self.columns): 50 -> min ratio to jump to reorder_labels: 1.0
#   len(self.columns): 55 -> min ratio to jump to reorder_labels: 1.0
#   ...
if rows_reordering_required and len(row_positions) * base_num_cols >= min(
    all_rows * len(self.columns) * base_ratio, len(row_positions) * base_num_cols
):
    # jump to reorder_labels
```

<details><summary>measurements with MODIN_CPUS=44</summary>

![image](https://github.com/modin-project/modin/assets/62142979/fd2df3f2-6e1b-478c-8602-57c986dbd550)

</details>

The reproducer from the issue (#6876) now performs like this:

![image](https://github.com/modin-project/modin/assets/62142979/79662ca8-637e-4e19-a508-ece6f5753e40)

- [x] first commit message and PR title follow format outlined [here](https://modin.readthedocs.io/en/latest/development/contributing.html#commit-message-formatting)
  > **_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` <!-- you can amend your commit with a signature via `git commit -amend -s` -->
- [ ] Resolves #6876 <!-- issue must be created for each patch -->
- [ ] tests added and passing
- [ ] module layout described at `docs/development/architecture.rst` is up-to-date <!-- if you have added, renamed or removed files or directories please update the documentation accordingly -->


Problem statement:
df.take is much slower against pandas
On a machine with 192 CPUs.

```python
# import pandas as pd
import modin.pandas as pd
import numpy as np
import time

df = pd.DataFrame(data=np.random.randint(99999, 99999999, size=(100000000,4)),
                columns=['C1','C2','C3','C4'])

to_take = np.random.randint(0, 100000000, size=80000000)
t0 = time.time()
df.take(to_take, axis=0)
t1 = time.time()
print('time to take: ', t1 - t0)
# time for take:  44.71011233329773 in Modin
# time for take:  3.082368850708008 in pandas
```

Hints from issue discussion:
cc @dchigarev 
```python
# import pandas as pd
import modin.pandas as pd
import numpy as np
import time

df = pd.DataFrame(data=np.random.randint(99999, 99999999, size=(100000000,1)),
                columns=['C1']).squeeze(axis=1)

to_take = np.random.randint(0, 100000000, size=80000000)
t0 = time.time()
df.take(to_take, axis=0)
t1 = time.time()
print('time to take: ', t1 - t0)
# time for take:  37.6995530128479 in Modin
# time for take:  1.492713212966919 in pandas
```

Commit message: PERF-#6876: Skip the masking stage on 'iloc' where beneficial

Signed-off-by: Dmitry Chigarev <dmitry.chigarev@intel.com>

Patch to review:
diff --git a/modin/core/dataframe/pandas/dataframe/dataframe.py b/modin/core/dataframe/pandas/dataframe/dataframe.py
index 7da33f52284..227b0e0a077 100644
--- a/modin/core/dataframe/pandas/dataframe/dataframe.py
+++ b/modin/core/dataframe/pandas/dataframe/dataframe.py
@@ -1161,12 +1161,22 @@ def _take_2d_positional(
                     extra_log="Mask takes only list-like numeric indexers, "
                     + f"received: {type(indexer)}",
                 )
+                if isinstance(indexer, list):
+                    indexer = np.array(indexer, dtype=np.int64)
             indexers.append(indexer)
         row_positions, col_positions = indexers
 
         if col_positions is None and row_positions is None:
             return self.copy()
 
+        # quite fast check that allows skip sorting
+        must_sort_row_pos = row_positions is not None and not np.all(
+            row_positions[1:] >= row_positions[:-1]
+        )
+        must_sort_col_pos = col_positions is not None and not np.all(
+            col_positions[1:] >= col_positions[:-1]
+        )
+
         if col_positions is None and row_positions is not None:
             # Check if the optimization that first takes part of the data using the mask
             # operation so that later less data is concatenated into a whole column is useful.
@@ -1175,18 +1185,40 @@ def _take_2d_positional(
             all_rows = None
             if self.has_materialized_index:
                 all_rows = len(self.index)
-            elif self._row_lengths_cache:
+            elif self._row_lengths_cache or must_sort_row_pos:
                 all_rows = sum(self._row_lengths_cache)
-            if all_rows:
-                if len(row_positions) > 0.9 * all_rows:
-                    return self._reorder_labels(
-                        row_positions=row_positions, col_positions=col_positions
-                    )
 
+            # 'base_num_cols' specifies the number of columns that the dataframe should have
+            # in order to jump to 'reordered_labels' in case of len(row_positions) / len(self) >= base_ratio;
+            # these variables may be a subject to change in order to tune performance more accurately
+            base_num_cols = 10
+            base_ratio = 0.2
+            # Example:
+            #   len(self.columns): 10 == base_num_cols -> min ratio to jump to reorder_labels: 0.2 == base_ratio
+            #   len(self.columns): 15 -> min ratio to jump to reorder_labels: 0.3
+            #   len(self.columns): 20 -> min ratio to jump to reorder_labels: 0.4
+            #   ...
+            #   len(self.columns): 49 -> min ratio to jump to reorder_labels: 0.98
+            #   len(self.columns): 50 -> min ratio to jump to reorder_labels: 1.0
+            #   len(self.columns): 55 -> min ratio to jump to reorder_labels: 1.0
+            #   ...
+            if (all_rows and len(row_positions) > 0.9 * all_rows) or (
+                must_sort_row_pos
+                and len(row_positions) * base_num_cols
+                >= min(
+                    all_rows * len(self.columns) * base_ratio,
+                    len(row_positions) * base_num_cols,
+                )
+            ):
+                return self._reorder_labels(
+                    row_positions=row_positions, col_positions=col_positions
+                )
         sorted_row_positions = sorted_col_positions = None
-
         if row_positions is not None:
-            sorted_row_positions = self._get_sorted_positions(row_positions)
+            if must_sort_row_pos:
+                sorted_row_positions = self._get_sorted_positions(row_positions)
+            else:
+                sorted_row_positions = row_positions
             # Get dict of row_parts as {row_index: row_internal_indices}
             row_partitions_dict = self._get_dict_of_block_index(
                 0, sorted_row_positions, are_indices_sorted=True
@@ -1201,7 +1233,10 @@ def _take_2d_positional(
             new_index = self.copy_index_cache(copy_lengths=True)
 
         if col_positions is not None:
-            sorted_col_positions = self._get_sorted_positions(col_positions)
+            if must_sort_col_pos:
+                sorted_col_positions = self._get_sorted_positions(col_positions)
+            else:
+                sorted_col_positions = col_positions
             # Get dict of col_parts as {col_index: col_internal_indices}
             col_partitions_dict = self._get_dict_of_block_index(
                 1, sorted_col_positions, are_indices_sorted=True


Code review comments:
diff_hunk: @@ -1175,18 +1185,40 @@ def _take_2d_positional(
             all_rows = None
             if self.has_materialized_index:
                 all_rows = len(self.index)
-            elif self._row_lengths_cache:
+            elif self._row_lengths_cache or must_sort_row_pos:
                 all_rows = sum(self._row_lengths_cache)
path: modin/core/dataframe/pandas/dataframe/dataframe.py
text: @user1:
Should we use `row_lengths` here?
```suggestion
                all_rows = sum(self.row_lengths)
```