Repository: modin-project/modin
Pull request: FIX-#4945: Fix `_take_2d_positional` that loses indexes due to filtering empty dataframes
Signed-off-by: Myachev <anatoly.myachev@intel.com>

<!--
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?

<!-- Please give a short brief about these changes. -->

- [x] commit message follows format outlined [here](https://modin.readthedocs.io/en/latest/development/contributing.html#commit-message-formatting)
- [x] passes `flake8 modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [x] passes `black --check modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [x] signed commit with `git commit -s` <!-- you can amend your commit with a signature via `git commit -amend -s` -->
- [x] Resolves #4945 <!-- issue must be created for each patch -->
- [x] tests added and passing
- [x] 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 -->
- [x] added (Issue Number: PR title (PR Number)) and github username to release notes for next major release <!-- e.g. DOCS-#4077: Add release notes template to docs folder (#4078) -->


Problem statement:
BUG: Lazily-evaluated DataFrame not computing Index properly
### System information
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: macOS 12.2.1
- **Modin version** (`modin.__version__`): 0.15.3
- **Python version**: 3.9.12
- **Code we can use to reproduce**:

```python
import modin.pandas as pd
import numpy as np
import decimal

df = pd.DataFrame(np.random.uniform(0.0,30.0,size=(60000,13))).add_prefix("col")
df1 = df[df['col0'] < 6.0].copy()

# This fails
df1['col0'] = df1['col0'].apply(lambda x: decimal.Decimal(str(x)))
```

### Describe the problem
`get_indices` returns an empty list instead of an empty `pandas.Index` which leads to a bug in certain cases.

### Source code / logs
<!-- Include any logs or source code that would be helpful to diagnose the problem. If including tracebacks, please include the full traceback. Large logs and files should be attached. Try to provide a reproducible test case that is the bare minimum necessary to generate the problem. -->

<details>
<summary>Error logs</summary>

```python
AttributeError                            Traceback (most recent call last)
Input In [6], in <cell line: 1>()
----> 1 df1['col0'] = df1['col0'].apply(lambda x: decimal.Decimal(str(x)))

File ~/Documents/modin/modin/logging/logger_decorator.py:128, in enable_logging.<locals>.decorator.<locals>.run_and_log(*args, **kwargs)
    113 """
    114 Compute function with logging if Modin logging is enabled.
    115 
   (...)
    125 Any
    126 """
    127 if LogMode.get() == "disable":
--> 128     return obj(*args, **kwargs)
    130 logger = get_logger()
    131 logger_level = getattr(logger, log_level)

File ~/Documents/modin/modin/pandas/dataframe.py:2517, in DataFrame.__setitem__(self, key, value)
   2515 if isinstance(value, Series):
   2516     value = value._query_compiler
-> 2517 self._update_inplace(self._query_compiler.setitem(0, key, value))

File ~/Documents/modin/modin/logging/logger_decorator.py:128, in enable_logging.<locals>.decorator.<locals>.run_and_log(*args, **kwargs)
    113 """
    114 Compute function with logging if Modin logging is enabled.
    115 
   (...)
    125 Any
    126 """
    127 if LogMode.get() == "disable":
--> 128     return obj(*args, **kwargs)
    130 logger = get_logger()
    131 logger_level = getattr(logger, log_level)

File ~/Documents/modin/modin/core/storage_formats/pandas/query_compiler.py:2234, in PandasQueryCompiler.setitem(self, axis, key, value)
   2233 def setitem(self, axis, key, value):
-> 2234     return self._setitem(axis=axis, key=key, value=value, how=None)

File ~/Documents/modin/modin/logging/logger_decorator.py:128, in enable_logging.<locals>.decorator.<locals>.run_and_log(*args, **kwargs)
    113 """
    114 Compute function with logging if Modin logging is enabled.
    115 
   (...)
    125 Any
    126 """
    127 if LogMode.get() == "disable":
--> 128     return obj(*args, **kwargs)
    130 logger = get_logger()
    131 logger_level = getattr(logger, log_level)

File ~/Documents/modin/modin/core/storage_formats/pandas/query_compiler.py:2296, in PandasQueryCompiler._setitem(self, axis, key, value, how)
   2294         value = value.transpose()
   2295     idx = self.get_axis(axis ^ 1).get_indexer_for([key])[0]
-> 2296     return self.insert_item(axis ^ 1, idx, value, how, replace=True)
   2298 # TODO: rework by passing list-like values to `apply_select_indices`
   2299 # as an item to distribute
   2300 if is_list_like(value):

File ~/Documents/modin/modin/logging/logger_decorator.py:128, in enable_logging.<locals>.decorator.<locals>.run_and_log(*args, **kwargs)
    113 """
    114 Compute function with logging if Modin logging is enabled.
    115 
   (...)
    125 Any
    126 """
    127 if LogMode.get() == "disable":
--> 128     return obj(*args, **kwargs)
    130 logger = get_logger()
    131 logger_level = getattr(logger, log_level)

File ~/Documents/modin/modin/core/storage_formats/base/query_compiler.py:3122, in BaseQueryCompiler.insert_item(self, axis, loc, value, how, replace)
   3120     second_mask_loc = loc + 1 if replace else loc
   3121     second_mask = mask(list(range(second_mask_loc, len(self.get_axis(axis)))))
-> 3122     return first_mask.concat(axis, [value, second_mask], join=how, sort=False)
   3123 else:
   3124     return self.concat(axis, [value], join=how, sort=False)

File ~/Documents/modin/modin/logging/logger_decorator.py:128, in enable_logging.<locals>.decorator.<locals>.run_and_log(*args, **kwargs)
    113 """
    114 Compute function with logging if Modin logging is enabled.
    115 
   (...)
    125 Any
    126 """
    127 if LogMode.get() == "disable":
--> 128     return obj(*args, **kwargs)
    130 logger = get_logger()
    131 logger_level = getattr(logger, log_level)

File ~/Documents/modin/modin/core/storage_formats/pandas/query_compiler.py:351, in PandasQueryCompiler.concat(self, axis, other, **kwargs)
    349 ignore_index = kwargs.get("ignore_index", False)
    350 other_modin_frame = [o._modin_frame for o in other]
--> 351 new_modin_frame = self._modin_frame.concat(axis, other_modin_frame, join, sort)
    352 result = self.__constructor__(new_modin_frame)
    353 if ignore_index:

File ~/Documents/modin/modin/logging/logger_decorator.py:128, in enable_logging.<locals>.decorator.<locals>.run_and_log(*args, **kwargs)
    113 """
    114 Compute function with logging if Modin logging is enabled.
    115 
   (...)
    125 Any
    126 """
    127 if LogMode.get() == "disable":
--> 128     return obj(*args, **kwargs)
    130 logger = get_logger()
    131 logger_level = getattr(logger, log_level)

File ~/Documents/modin/modin/core/dataframe/pandas/dataframe/dataframe.py:124, in lazy_metadata_decorator.<locals>.decorator.<locals>.run_f_on_minimally_updated_metadata(self, *args, **kwargs)
    122     elif apply_axis == "rows":
    123         obj._propagate_index_objs(axis=0)
--> 124 result = f(self, *args, **kwargs)
    125 if apply_axis is None and not transpose:
    126     result._deferred_index = self._deferred_index

File ~/Documents/modin/modin/core/dataframe/pandas/dataframe/dataframe.py:2786, in PandasDataframe.concat(self, axis, others, how, sort)
   2779     new_widths = _compute_new_widths()
   2780 else:
   2781     (
   2782         left_parts,
   2783         right_parts,
   2784         joined_index,
   2785         partition_sizes_along_axis,
-> 2786     ) = self._copartition(
   2787         axis.value ^ 1, others, how, sort, force_repartition=False
   2788     )
   2789     if axis == Axis.COL_WISE:
   2790         new_lengths = partition_sizes_along_axis

File ~/Documents/modin/modin/logging/logger_decorator.py:128, in enable_logging.<locals>.decorator.<locals>.run_and_log(*args, **kwargs)
    113 """
    114 Compute function with logging if Modin logging is enabled.
    115 
   (...)
    125 Any
    126 """
    127 if LogMode.get() == "disable":
--> 128     return obj(*args, **kwargs)
    130 logger = get_logger()
    131 logger_level = getattr(logger, log_level)

File ~/Documents/modin/modin/core/dataframe/pandas/dataframe/dataframe.py:2571, in PandasDataframe._copartition(self, axis, other, how, sort, force_repartition)
   2569 self_index = self.axes[axis]
   2570 others_index = [o.axes[axis] for o in other]
-> 2571 joined_index, make_reindexer = self._join_index_objects(
   2572     axis, [self_index] + others_index, how, sort
   2573 )
   2575 frames = [self] + other
   2576 non_empty_frames_idx = [
   2577     i for i, o in enumerate(frames) if o._partitions.size != 0
   2578 ]

File ~/Documents/modin/modin/logging/logger_decorator.py:128, in enable_logging.<locals>.decorator.<locals>.run_and_log(*args, **kwargs)
    113 """
    114 Compute function with logging if Modin logging is enabled.
    115 
   (...)
    125 Any
    126 """
    127 if LogMode.get() == "disable":
--> 128     return obj(*args, **kwargs)
    130 logger = get_logger()
    131 logger_level = getattr(logger, log_level)

File ~/Documents/modin/modin/core/dataframe/pandas/dataframe/dataframe.py:1445, in PandasDataframe._join_index_objects(axis, indexes, how, sort)
   1442         return left_index.join(right_index, how=how, sort=sort)
   1444 # define condition for joining indexes
-> 1445 all_indices_equal = all(indexes[0].equals(index) for index in indexes[1:])
   1446 do_join_index = how is not None and not all_indices_equal
   1448 # define condition for joining indexes with getting indexers

File ~/Documents/modin/modin/core/dataframe/pandas/dataframe/dataframe.py:1445, in <genexpr>(.0)
   1442         return left_index.join(right_index, how=how, sort=sort)
   1444 # define condition for joining indexes
-> 1445 all_indices_equal = all(indexes[0].equals(index) for index in indexes[1:])
   1446 do_join_index = how is not None and not all_indices_equal
   1448 # define condition for joining indexes with getting indexers

AttributeError: 'list' object has no attribute 'equals'
```

</details>

Hints from issue discussion:
@anmyachev I've been digging into this bug today and it seems that the root issue is from your PR here: https://github.com/modin-project/modin/pull/4921. Although, we currently are getting a different error thrown for the reproducer, I'm finding that we are returning the incorrect result even after resolving the initial error. 

I first thought the issue was with how `copy` doesn't recompute the `index` if it's None, but the problem was deeper. 
We can't just set `new_index` to `index_cache` in `_take_2d_positional` because there may be cases (like the one above) where the `partitions` are empty, and therefore the `index` cannot be recomputed at a later stage. I don't know if this affects any of the other metadata, but it's causing subtle failures like the case above. I also don't know if the true bug is the fact that `new_partitions` is empty, triggering the computation of an empty Index. Any insight would be much appreciated!

cc: @mvashishtha 
We are gradually moving towards asynchronous calculations, and my pull request is a logical continuation of the fact that indexes are no longer always calculated immediately.

It seems to me that the main problem here is that we are filtering out dataframes that are empty, which results in the list of partitions being empty. However, this means that we lose the metadata contained in empty dataframes. As far as I could see, basically these leads to the loss of the correct `dtypes`.

Example:
```python
>>> df = pd.DataFrame({"col1": [1, 2, 3, 4], "col2": [1.3, 2.5, 2.9, 1.6]})
>>> df
   col1  col2
0     1   1.3
1     2   2.5
2     3   2.9
3     4   1.6
>>> df.dtypes
col1      int64
col2    float64
dtype: object
>>> df[df["col1"] < 0]
Empty DataFrame
Columns: [col1, col2]
Index: []
>>> df[df["col1"] < 0].dtypes
col1      int64
col2    float64
dtype: object
>>> df[df["col1"] < 0].empty
True
```

Thanks @anmyachev . Do you have any ideas on what we can do here? Also I'm not sure if the filter in my example is creating any empty dfs. They shouldn't be empty. I can update the reproducer to show you what I'm using now. Updated reproducer:

```python
import modin.pandas as pd
import numpy as np

df = pd.DataFrame({"col0": [0,1]})
df1 = df[df['col0'] < 6.0].copy()

# This returns an empty dataframe
df1['col0'] = df1['col0'].apply(lambda x: x+1)
```
@pyrito I will take a look at this.
@anmyachev if it helps, I've now learned that the bug only shows up because of this PR: https://github.com/modin-project/modin/pull/4911

Commit message: FIX-#4945: fix _take_2d_positional that missed indexes due to filtering

Signed-off-by: Myachev <anatoly.myachev@intel.com>

Patch to review:
diff --git a/modin/core/dataframe/pandas/dataframe/dataframe.py b/modin/core/dataframe/pandas/dataframe/dataframe.py
index 9a3f7c363c7..d3c2ecef5ca 100644
--- a/modin/core/dataframe/pandas/dataframe/dataframe.py
+++ b/modin/core/dataframe/pandas/dataframe/dataframe.py
@@ -852,12 +852,8 @@ def _take_2d_positional(
                         row_internal_indices, col_internal_indices
                     )
                     for col_idx, col_internal_indices in col_partitions_dict.items()
-                    if isinstance(col_internal_indices, slice)
-                    or len(col_internal_indices) > 0
                 ]
                 for row_idx, row_internal_indices in row_partitions_dict.items()
-                if isinstance(row_internal_indices, slice)
-                or len(row_internal_indices) > 0
             ]
         )
         intermediate = self.__constructor__(
@@ -1343,6 +1339,9 @@ def _get_dict_of_block_index(self, axis, indices, are_indices_sorted=False):
         if isinstance(indices, list):
             # Converting python list to numpy for faster processing
             indices = np.array(indices, dtype=np.int64)
+        # Fasttrack empty numpy array
+        if isinstance(indices, np.ndarray) and indices.size == 0:
+            return OrderedDict([(0, np.array([], dtype=np.int64))])
         negative_mask = np.less(indices, 0)
         has_negative = np.any(negative_mask)
         if has_negative:
diff --git a/modin/core/dataframe/pandas/partitioning/partition_manager.py b/modin/core/dataframe/pandas/partitioning/partition_manager.py
index dcf19f6526d..442cb083022 100644
--- a/modin/core/dataframe/pandas/partitioning/partition_manager.py
+++ b/modin/core/dataframe/pandas/partitioning/partition_manager.py
@@ -1451,7 +1451,6 @@ def rebalance_partitions(cls, partitions):
                     for obj in partitions[stop]:
                         obj._length_cache = new_last_partition_size
 
-                    partition_size = ideal_partition_size
                 # The new virtual partitions are not `full_axis`, even if they
                 # happen to span all rows in the dataframe, because they are
                 # meant to be the final partitions of the dataframe. They've


Code review comments:
diff_hunk: @@ -1451,7 +1451,6 @@ def rebalance_partitions(cls, partitions):
                     for obj in partitions[stop]:
                         obj._length_cache = new_last_partition_size
 
-                    partition_size = ideal_partition_size
path: modin/core/dataframe/pandas/partitioning/partition_manager.py
text: @author:
Just extra assignment, not directly related to this pull request.

@user1:
Can we revert this change if it's not relevant to the PR?