Repository: modin-project/modin
Pull request: [FIX] Fix of inconsistent indices #1726 #1731 #1732 #1734
<!--
Thank you for your contribution! 
Please review the contributing docs: https://modin.readthedocs.io/en/latest/contributing.html
if you have questions about contributing.
-->

## What do these changes do?

<!-- Please give a short brief about these changes. -->

- [x] passes `flake8 modin`
- [x] passes `black --check modin`
- [x] signed commit with `git commit -s` <!-- you can amend your commit with a signature via `git commit -amend -s` -->
- [x] Resolves #1726, #1731, #1732 and #1734 <!-- issue must be created for each patch -->
- [x] tests added and passing

Initially, this PR was fixing that `to_pandas` in pandas backend does not consider information about indices/columns of modin frame, only from partitions. But after that fix many tests started to fail because `df_equals`, that uses in many tests, use `to_pandas` to convert modin frame to pandas and then compare it with an expected result. It revealed that many operations returns modin frame with incorrect indices, so this PR is also fixing all of them.

Problem statement:
to_pandas in PandasQueryCompiler loses meta information about indices and columns
### System information
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: Windows 10
- **Modin version** (`modin.__version__`): 0.7.3+198.ge4e3fecb
- **Python version**: 3.7.5
- **Code we can use to reproduce**:
```python
if __name__ == "__main__":
    import modin.pandas as pd
    import numpy as np

    data = {"A": np.arange(256)}

    index = pd.MultiIndex.from_tuples((i, i*2) for i in np.arange(257)).drop(0)

    md_df = pd.DataFrame(data, index=index)

    print("md_df.index:", len(md_df.index.levels[0]))  # md_df.index: 257 
    print("pd_df.index:", len(md_df._to_pandas().index.levels[0]))  # pd_df.index: 256

```
<!--
You can obtain the Modin version with

python -c "import modin; print(modin.__version__)"
-->

### Describe the problem
<!-- Describe the problem clearly here. -->
`to_pandas` loses information about index levels which is useful for operations that operates with indices levels internally (like `unstack`)


Inconsistent columns in groupby_reduce result
### System information
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: Windows 10
- **Modin version** (`modin.__version__`): 0.7.3+200.g8e7a5682
- **Python version**: 3.7.5
- **Code we can use to reproduce**:
```python
if __name__ == "__main__":
    import modin.pandas as pd
    import pandas
    import numpy as np

    data = {
        "col1": [0, 1, 2, 3],
        "col2": [4, 5, np.NaN, 7],
        "col3": [np.NaN, np.NaN, 12, 10],
        "col4": [17, 13, 16, 15],
        "col5": [-4, -5, -6, -7],
    }

    md_df, pd_df = pd.DataFrame(data), pandas.DataFrame(data)

    groupby_kwargs = {"by": ["col5", "col4", "col1"], "as_index": False}

    md_result, pd_result = (
        md_df.groupby(**groupby_kwargs).any(),
        pd_df.groupby(**groupby_kwargs).any(),
    )

    print("pd_result:\n", pd_result, sep="")
    print("\nmd_result:\n", md_result, sep="")

    print("\npd_columns:", pd_result.columns)
    print("md_columns:", md_result.columns)

```

<details><summary>Output:</summary>


```
pd_result:
   col5  col4  col1   col2   col3
0    -7    15     3   True   True
1    -6    16     2  False   True
2    -5    13     1   True  False
3    -4    17     0   True  False

md_result:
   col5  col4  col1   col2   col3
0    -7    15     3   True   True
1    -6    16     2  False   True
2    -5    13     1   True  False
3    -4    17     0   True  False

pd_columns: Index(['col5', 'col4', 'col1', 'col2', 'col3'], dtype='object')
md_columns: Index(['col1', 'col2', 'col3', 'col4', 'col5'], dtype='object')
```


</details>


<!--
You can obtain the Modin version with

python -c "import modin; print(modin.__version__)"
-->



### Describe the problem
Columns into partitions seems to be correct, but columns of dataframe itself isn't, that's also the reason why test don't fails on that test case (`to_pandas` that used in `df_equals` considers information only from partitions)


Incorrect columns in the result of binary operations
### System information
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: Windows 10
- **Modin version** (`modin.__version__`): 0.7.3+200.g8e7a5682
- **Python version**: 3.7.5
- **Code we can use to reproduce**:
```python
if __name__ == "__main__":
    import modin.pandas as pd
    import pandas
    import numpy as np

    data = {key: np.arange(i, i + 5) for i, key in enumerate(["C", "B", "A"])}

    md_df, pd_df = pd.DataFrame(data), pandas.DataFrame(data)

    pd_res = pd_df + pd_df.iloc[0, :-1]
    md_res = md_df + md_df.iloc[0, :-1]

    print("pd_result:\n", pd_res, sep="")
    print("\nmd_result:\n", md_res, sep="")

    print("\npd_columns:", pd_res.columns)
    print("md_columns:", md_res.columns)

```

<details><summary>Output:</summary>


```
pd_result:
    A    B    C
0 NaN  2.0  0.0
1 NaN  3.0  1.0
2 NaN  4.0  2.0
3 NaN  5.0  3.0
4 NaN  6.0  4.0

md_result:
    A    B    C
0 NaN  2.0  0.0
1 NaN  3.0  1.0
2 NaN  4.0  2.0
3 NaN  5.0  3.0
4 NaN  6.0  4.0

pd_columns: Index(['A', 'B', 'C'], dtype='object')
md_columns: Index(['C', 'B', 'A'], dtype='object')
```


</details>



<!--
You can obtain the Modin version with

python -c "import modin; print(modin.__version__)"
-->

### Describe the problem
Columns into partitions seems to be correct, but columns of dataframe itself isn't, that's also the reason why test don't fails on that test case (`to_pandas` that used in `df_equals` considers information only from partitions)


Incorrect result of nlargest function
### System information
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: Windows 10
- **Modin version** (`modin.__version__`): 0.7.3+200.g8e7a5682
- **Python version**: 3.7.5
- **Code we can use to reproduce**:
```python
if __name__ == "__main__":
    import modin.pandas as pd
    import pandas
    
     # test case from test_nlargest
    data = {
            "population": [
                59000000,
                65000000,
                434000,
                434000,
                434000,
                337000,
                11300,
                11300,
                11300,
            ],
            "GDP": [1937894, 2583560, 12011, 4520, 12128, 17036, 182, 38, 311],
            "alpha-2": ["IT", "FR", "MT", "MV", "BN", "IS", "NR", "TV", "AI"],
        }
    index = [
        "Italy",
        "France",
        "Malta",
        "Maldives",
        "Brunei",
        "Iceland",
        "Nauru",
        "Tuvalu",
        "Anguilla",
    ]
    modin_df = pd.DataFrame(data=data, index=index)
    pandas_df = pandas.DataFrame(data=data, index=index)

    md_res = modin_df.nlargest(3, "population")
    pd_res = pandas_df.nlargest(3, "population")

    print("pd_res:\n", pd_res, sep="")
    print("\nmd_res_partitions:\n", md_res._query_compiler._modin_frame._partitions[0][0].to_pandas(), sep="")
    print("\nmd_res:\n", md_res, sep="") # Exception: IndexError('positional indexers are out-of-bounds')

```

<details><summary>Output:</summary>


```
pd_res:
        population      GDP alpha-2
France    65000000  2583560      FR
Italy     59000000  1937894      IT
Malta       434000    12011      MT

md_res_partitions:
        population      GDP alpha-2
France    65000000  2583560      FR
Italy     59000000  1937894      IT
Malta       434000    12011      MT

md_res:
distributed.worker - WARNING -  Compute Failed
Function:  apply_list_of_funcs
args:      ([[b"\x80\x04\x95\xb0\x03\x00\x00\x00\x00\x00\x00\x8c\x17cloudpickle.cloudpickle\x94\x8c\x0e_fill_function\x94\x93\x94(h\x00\x8c\x0f_make_skel_func\x94\x93\x94h\x00\x8c\r_builtin_type\x94\x93\x94\x8c\x08CodeType\x94\x85\x94R\x94(K\x01K\x00K\x01K\x05K\x13C\x14t\x00\xa0\x01|\x00j\x02\x88\x01\x88\x00f\x02\x19\x00\xa1\x01S\x00\x94N\x85\x94\x8c\x06pandas\x94\x8c\tDataFrame\x94\x8c\x04iloc\x94\x87\x94\x8c\x02df\x94\x85\x94\x8cZc:\\users\\dchigare\\desktop\\repos\\modin\\modin\\engines\\dask\\pandas_on_dask\\frame\\partition.py\x94\x8c\x08<lambda>\x94KcC\x00\x94\x8c\x0bcol_indices\x94\x8c\x0brow_indices\x94\x86\x94)t\x94R\x94K\x02}\x94(\x8c\x0b__package__\x94\x8c'modin.engines.dask.pandas_on_dask.frame\x94\x8c\x08__name__\x94\x8c1modin.engines.dask.pandas_on_dask.frame.partition\x94\x8c\x08__file__\x94h\x12u\x87\x94R\x94}\x94(\x8c\x07globals\x94}\x94h\x0ch\x00\x8c\tsubimport\x94\x93\x94h\x0c\x85\x94R\x94s\x8c\x08defaults\x94N\x8c\x04dict\x94}\x94\x8c\x0eclosure_values\x94]\x94(\x8c\x15numpy.c
kwargs:    {}
Exception: IndexError('positional indexers are out-of-bounds')

Traceback (most recent call last):
  File "C:\Users\dchigare\Desktop\REPOS\TESTS\reprod.py", line 39, in <module>
    print("\nmd_res:\n", md_res, sep="") # Exception: IndexError('positional indexers are out-of-bounds')
  File "c:\users\dchigare\desktop\repos\modin\modin\pandas\base.py", line 3465, in __str__
    return repr(self)
  File "c:\users\dchigare\desktop\repos\modin\modin\pandas\dataframe.py", line 162, in __repr__
    result = repr(self._build_repr_df(num_rows, num_cols))
  File "c:\users\dchigare\desktop\repos\modin\modin\pandas\base.py", line 100, in _build_repr_df
    return self.iloc[indexer]._query_compiler.to_pandas()
  File "c:\users\dchigare\desktop\repos\modin\modin\backends\pandas\query_compiler.py", line 188, in to_pandas
    return self._modin_frame.to_pandas()
  File "c:\users\dchigare\desktop\repos\modin\modin\engines\base\frame\data.py", line 1350, in to_pandas
    df = self._frame_mgr_cls.to_pandas(self._partitions)
  File "c:\users\dchigare\desktop\repos\modin\modin\engines\base\frame\partition_manager.py", line 258, in to_pandas
    retrieved_objects = [[obj.to_pandas() for obj in part] for part in partitions]
  File "c:\users\dchigare\desktop\repos\modin\modin\engines\base\frame\partition_manager.py", line 258, in <listcomp>
    retrieved_objects = [[obj.to_pandas() for obj in part] for part in partitions]
  File "c:\users\dchigare\desktop\repos\modin\modin\engines\base\frame\partition_manager.py", line 258, in <listcomp>
    retrieved_objects = [[obj.to_pandas() for obj in part] for part in partitions]
  File "c:\users\dchigare\desktop\repos\modin\modin\engines\dask\pandas_on_dask\frame\partition.py", line 127, in to_pandas
    dataframe = self.get()
  File "c:\users\dchigare\desktop\repos\modin\modin\engines\dask\pandas_on_dask\frame\partition.py", line 63, in get
    return self.future.result()
  File "C:\Program Files (x86)\Microsoft Visual Studio\Shared\Python37_64\lib\site-packages\distributed\client.py", line 218, in result
    raise exc.with_traceback(tb)
  File "c:\users\dchigare\desktop\repos\modin\modin\engines\dask\pandas_on_dask\frame\partition.py", line 27, in apply_list_of_funcs
    df = func(df, **kwargs)
  File "c:\users\dchigare\desktop\repos\modin\modin\engines\dask\pandas_on_dask\frame\partition.py", line 99, in <lambda>
    lambda df: pandas.DataFrame(df.iloc[row_indices, col_indices])
  File "C:\Program Files (x86)\Microsoft Visual Studio\Shared\Python37_64\lib\site-packages\pandas\core\indexing.py", line 1762, in __getitem__
    return self._getitem_tuple(key)
  File "C:\Program Files (x86)\Microsoft Visual Studio\Shared\Python37_64\lib\site-packages\pandas\core\indexing.py", line 2067, in _getitem_tuple
    self._has_valid_tuple(tup)
  File "C:\Program Files (x86)\Microsoft Visual Studio\Shared\Python37_64\lib\site-packages\pandas\core\indexing.py", line 703, in _has_valid_tuple
    self._validate_key(k, i)
  File "C:\Program Files (x86)\Microsoft Visual Studio\Shared\Python37_64\lib\site-packages\pandas\core\indexing.py", line 2009, in _validate_key
    raise IndexError("positional indexers are out-of-bounds")
IndexError: positional indexers are out-of-bounds
```


</details>




<!--
You can obtain the Modin version with

python -c "import modin; print(modin.__version__)"
-->

### Describe the problem
If we print `columns` and `index` of `md_res` in debug, we can see that it contains incorrect values. But as we see from the output of above code, partitions itself contains the correct result, that's actually the reason why that tests don't fail in CI (`df_equals` that used in tests converts modin result via `to_pandas` which consider information only from partitions)
```
(Pdb) md_res.columns
Index(['__reduced__'], dtype='object')
(Pdb) md_res.index
Index(['Italy', 'France', 'Malta', 'Maldives', 'Brunei', 'Iceland', 'Nauru',
       'Tuvalu', 'Anguilla'],
      dtype='object')
```

Hints from issue discussion:
I figured out that `to_pandas` does not consider indices and columns from modin_frame at all, only from partitions. For example if modin frame has `columns = ['A', 'B']` but in partitions `columns = [1, 2, 3]` (different columns lengths could appear because of incorrect function implementation) then `to_pandas` will return us the DataFrame with `columns = [1, 2, 3]`.

More explicit example why it is important:
Many tests in modin uses [`df_equals`](https://github.com/modin-project/modin/blob/master/modin/pandas/test/utils.py#L396) to compare `modin_result` with `pandas_result`. Under the hood `df_equals` [translates `modin_df` to pandas](https://github.com/modin-project/modin/blob/master/modin/pandas/test/utils.py#L444-L448) and after that compares its like two pandas dataframes. Since `to_pandas` does not consider modin frame indices, we will never know if modin frame itself has correct indices/columns in such comparison.

For example, let's take `nlargest` function, its tests pass ok, but when we would like to use its result without converting it to pandas, we will see that it's incorrect (#1734).

Many tests started to fail after I fixed that `to_pandas` does not use modin frame indices in the conversion to pandas (#1729 #1731 #1732 #1733 #1734), I've fixed most of them in #1727.

@devin-petersohn please take a look, I guess that all the PRs with new pandas API functions implementation should be blocked for now, because it can't be properly tested

Commit message: FIX-#1726: make to_pandas consider external labels

This fix is also fixing inconsistent indices in some of operations (#1731, #1732 and #1734)

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

Patch to review:
diff --git a/modin/backends/pandas/query_compiler.py b/modin/backends/pandas/query_compiler.py
index 14f12ff1a2c..57ce0f06109 100644
--- a/modin/backends/pandas/query_compiler.py
+++ b/modin/backends/pandas/query_compiler.py
@@ -468,13 +468,13 @@ def transpose(self, *args, **kwargs):
 
     # MapReduce operations
 
-    def _is_monotonic(self, type=None):
+    def _is_monotonic(self, func_type=None):
         funcs = {
             "increasing": lambda df: df.is_monotonic_increasing,
             "decreasing": lambda df: df.is_monotonic_decreasing,
         }
 
-        monotonic_fn = funcs.get(type, funcs["increasing"])
+        monotonic_fn = funcs.get(func_type, funcs["increasing"])
 
         def is_monotonic_map(df):
             df = df.squeeze(axis=1)
@@ -494,10 +494,12 @@ def is_monotonic_reduce(df):
             edge_case = monotonic_fn(pandas.Series(edges_list))
             return [common_case and edge_case]
 
-        return MapReduceFunction.register(is_monotonic_map, is_monotonic_reduce)(self)
+        return MapReduceFunction.register(
+            is_monotonic_map, is_monotonic_reduce, axis=0
+        )(self)
 
     def is_monotonic_decreasing(self):
-        return self._is_monotonic(type="decreasing")
+        return self._is_monotonic(func_type="decreasing")
 
     is_monotonic = _is_monotonic
 
@@ -626,7 +628,6 @@ def sort_index_for_equal_values(result, ascending):
     idxmin = ReductionFunction.register(pandas.DataFrame.idxmin)
     median = ReductionFunction.register(pandas.DataFrame.median)
     nunique = ReductionFunction.register(pandas.DataFrame.nunique)
-    nlargest = ReductionFunction.register(pandas.DataFrame.nlargest)
     skew = ReductionFunction.register(pandas.DataFrame.skew)
     kurt = ReductionFunction.register(pandas.DataFrame.kurt)
     std = ReductionFunction.register(pandas.DataFrame.std)
@@ -1162,13 +1163,17 @@ def map_func(df, other=other, squeeze_self=squeeze_self):
         )
         return self.__constructor__(new_modin_frame)
 
-    def nsmallest(self, n, columns=None, keep="first"):
+    def nsort(self, n, columns=None, keep="first", sort_type="nsmallest"):
         def map_func(df, n=n, keep=keep, columns=columns):
             if columns is None:
                 return pandas.DataFrame(
-                    pandas.Series.nsmallest(df.squeeze(axis=1), n=n, keep=keep)
+                    getattr(pandas.Series, sort_type)(
+                        df.squeeze(axis=1), n=n, keep=keep
+                    )
                 )
-            return pandas.DataFrame.nsmallest(df, n=n, columns=columns, keep=keep)
+            return getattr(pandas.DataFrame, sort_type)(
+                df, n=n, columns=columns, keep=keep
+            )
 
         if columns is None:
             new_columns = ["__reduced__"]
@@ -1180,6 +1185,12 @@ def map_func(df, n=n, keep=keep, columns=columns):
         )
         return self.__constructor__(new_modin_frame)
 
+    def nsmallest(self, *args, **kwargs):
+        return self.nsort(sort_type="nsmallest", *args, **kwargs)
+
+    def nlargest(self, *args, **kwargs):
+        return self.nsort(sort_type="nlargest", *args, **kwargs)
+
     def eval(self, expr, **kwargs):
         """Returns a new QueryCompiler with expr evaluated on columns.
 
diff --git a/modin/data_management/functions/binary_function.py b/modin/data_management/functions/binary_function.py
index 5933b5b4283..b8edf8f608a 100644
--- a/modin/data_management/functions/binary_function.py
+++ b/modin/data_management/functions/binary_function.py
@@ -39,6 +39,7 @@ def caller(query_compiler, other, *args, **kwargs):
                             axis,
                             lambda l, r: func(l, r.squeeze(), *args, **kwargs),
                             other._modin_frame,
+                            preserve_labels=call_kwds.get("preserve_labels", False),
                         )
                     )
                 else:
diff --git a/modin/data_management/functions/groupby_function.py b/modin/data_management/functions/groupby_function.py
index a1801e17105..6d2373d83bf 100644
--- a/modin/data_management/functions/groupby_function.py
+++ b/modin/data_management/functions/groupby_function.py
@@ -115,19 +115,9 @@ def compute_reduce(df):
                 except ValueError:
                     return compute_reduce(df.copy())
 
-            if axis == 0:
-                new_columns = qc.columns
-                new_index = None
-            else:
-                new_index = query_compiler.index
-                new_columns = None
+            # TODO: try to precompute `new_index` and `new_columns`
             new_modin_frame = qc._modin_frame.groupby_reduce(
-                axis,
-                by._modin_frame,
-                _map,
-                _reduce,
-                new_columns=new_columns,
-                new_index=new_index,
+                axis, by._modin_frame, _map, _reduce
             )
             return query_compiler.__constructor__(new_modin_frame)
 
diff --git a/modin/engines/base/frame/data.py b/modin/engines/base/frame/data.py
index 520e9d28bd8..907c08fe485 100644
--- a/modin/engines/base/frame/data.py
+++ b/modin/engines/base/frame/data.py
@@ -16,6 +16,7 @@
 import pandas
 from pandas.core.indexes.api import ensure_index
 from pandas.core.dtypes.common import is_numeric_dtype
+from typing import Union
 
 from modin.backends.pandas.query_compiler import PandasQueryCompiler
 from modin.error_message import ErrorMessage
@@ -40,18 +41,22 @@ def __init__(
         row_lengths=None,
         column_widths=None,
         dtypes=None,
+        validate_axes: Union[bool, str] = False,
     ):
         """Initialize a dataframe.
 
-        Args:
-            partitions: A 2D NumPy array of partitions. Must contain partition objects.
-            index: The index object for the dataframe. Converts to a pandas.Index.
-            columns: The columns object for the dataframe. Converts to a pandas.Index.
-            row_lengths: (optional) The lengths of each partition in the rows. The
+        Parameters
+        ----------
+            partitions : A 2D NumPy array of partitions. Must contain partition objects.
+            index : The index object for the dataframe. Converts to a pandas.Index.
+            columns : The columns object for the dataframe. Converts to a pandas.Index.
+            row_lengths : (optional) The lengths of each partition in the rows. The
                 "height" of each of the block partitions. Is computed if not provided.
-            column_widths: (optional) The width of each partition in the columns. The
+            column_widths : (optional) The width of each partition in the columns. The
                 "width" of each of the block partitions. Is computed if not provided.
-            dtypes: (optional) The data types for the dataframe.
+            dtypes : (optional) The data types for the dataframe.
+            validate_axes : (optional) Whether or not validate for equality
+                internal indices of partitions and passed `index` and `columns`.
         """
         self._partitions = partitions
         self._index_cache = ensure_index(index)
@@ -74,6 +79,8 @@ def __init__(
         self._column_widths_cache = column_widths
         self._dtypes = dtypes
         self._filter_empties()
+        if validate_axes is not False:
+            self._validate_internal_indices(mode=validate_axes)
 
     @property
     def _row_lengths(self):
@@ -204,6 +211,21 @@ def _set_columns(self, new_columns):
                 self._dtypes.index = new_columns
         self._apply_index_objs(axis=1)
 
+    def _set_axis(self, axis, new_axis):
+        """Replaces the current labels at the specified axis with the new one
+
+        Parameters
+        ----------
+            axis : int,
+                Axis to set labels along
+            new_axis : Index,
+                The replacement labels
+        """
+        if axis:
+            self._set_columns(new_axis)
+        else:
+            self._set_index(new_axis)
+
     columns = property(_get_columns, _set_columns)
     index = property(_get_index, _set_index)
 
@@ -234,6 +256,74 @@ def _filter_empties(self):
         self._column_widths_cache = [w for w in self._column_widths if w > 0]
         self._row_lengths_cache = [r for r in self._row_lengths if r > 0]
 
+    def _validate_axis_equality(self, axis: int, update: bool):
+        """
+        Validates internal and external indices of modin_frame at the specified axis.
+
+        Parameters
+        ----------
+            axis : int,
+                Axis to validate indices along
+            update : bool,
+                Update or raise an exception if indices don't match
+        """
+        internal_axis = self._frame_mgr_cls.get_indices(
+            axis, self._partitions, lambda df: df.axes[axis]
+        )
+        is_equals = self.axes[axis].equals(internal_axis)
+        if not update:
+            ErrorMessage.catch_bugs_and_request_email(
+                not is_equals, "Internal and external indices do not match.",
+            )
+        elif not is_equals:
+            self._set_axis(axis, self.axes[axis])
+
+    def _validate_internal_indices(self, mode=None, **kwargs):
+        """
+        Validates and optionally updates internal and external indices
+        of modin_frame in specified mode. There is 3 modes supported:
+            1. "reduced" - validates and updates indices on that axes
+                where external indices is ["__reduced__"]
+            2. "all" - validates indices at all axes, optionally updates
+                internal indices if `update` parameter specified in kwargs
+            3. "custom" - validation follows arguments specified in kwargs.
+
+        Parameters
+        ----------
+            mode : str or bool, default None
+            validate_index : bool, (optional, could be specified via `mode`)
+            validate_columns : bool, (optional, could be specified via `mode`)
+            update : bool, (optional, could be specified via `mode`)
+        """
+
+        if isinstance(mode, bool):
+            default_update_value = kwargs.get("update", mode)
+            mode = "all"
+        else:
+            default_update_value = kwargs.get("update", False)
+
+        reduced_sample = pandas.Index(["__reduced__"])
+        args_dict = {
+            "custom": kwargs,
+            "reduced": {
+                "validate_index": self.index.equals(reduced_sample),
+                "validate_columns": self.columns.equals(reduced_sample),
+                "update": True,
+            },
+            "all": {
+                "validate_index": True,
+                "validate_columns": True,
+                "update": default_update_value,
+            },
+        }
+
+        args = args_dict.get(mode, args_dict["custom"])
+
+        if args.get("validate_index", True):
+            self._validate_axis_equality(0, args.get("update", False))
+        if args.get("validate_columns", True):
+            self._validate_axis_equality(1, args.get("update", False))
+
     def _apply_index_objs(self, axis=None):
         """Lazily applies the index object (Index or Columns) to the partitions.
 
@@ -872,7 +962,13 @@ def _compute_map_reduce_metadata(self, axis, new_parts):
             else:
                 new_dtypes = self._dtypes
         return self.__constructor__(
-            new_parts, index, columns, new_lengths, new_widths, new_dtypes
+            new_parts,
+            index,
+            columns,
+            new_lengths,
+            new_widths,
+            new_dtypes,
+            validate_axes="reduced",
         )
 
     def _fold_reduce(self, axis, func):
@@ -936,7 +1032,9 @@ def _map_reduce(self, axis, map_func, reduce_func=None, preserve_index=True):
                     0, reduce_parts, lambda df: df.index
                 )
                 new_columns = ["__reduced__"]
-            return self.__constructor__(reduce_parts, new_index, new_columns)
+            return self.__constructor__(
+                reduce_parts, new_index, new_columns, validate_axes="reduced"
+            )
 
     def _map(self, func, dtypes=None, validate_index=False):
         """Perform a function that maps across the entire dataset.
@@ -1095,7 +1193,13 @@ def _apply_full_axis(
                 [np.dtype(dtypes)] * len(new_columns), index=new_columns
             )
         return self.__constructor__(
-            new_partitions, new_index, new_columns, None, None, dtypes
+            new_partitions,
+            new_index,
+            new_columns,
+            None,
+            None,
+            dtypes,
+            validate_axes="reduced",
         )
 
     def _apply_full_axis_select_indices(
@@ -1247,7 +1351,6 @@ def broadcast_apply(self, axis, func, other, preserve_labels=True, dtypes=None):
         Returns:
              A new Modin DataFrame
         """
-        assert preserve_labels, "`preserve_labels=False` Not Yet Implemented"
         # Only sort the indices if they do not match
         left_parts, right_parts, joined_index = self._copartition(
             axis, other, "left", sort=not self.axes[axis].equals(other.axes[axis])
@@ -1261,6 +1364,11 @@ def broadcast_apply(self, axis, func, other, preserve_labels=True, dtypes=None):
             dtypes = self._dtypes
         new_index = self.index
         new_columns = self.columns
+        if not preserve_labels:
+            if axis == 1:
+                new_columns = joined_index
+            else:
+                new_index = joined_index
         return self.__constructor__(
             new_frame, new_index, new_columns, None, None, dtypes=dtypes
         )
@@ -1476,7 +1584,13 @@ def to_pandas(self):
                 df = pandas.DataFrame(columns=self.columns)
             else:
                 df = pandas.DataFrame(columns=self.columns, index=self.index)
-        df.index.name = self.index.name
+        else:
+            ErrorMessage.catch_bugs_and_request_email(
+                not df.index.equals(self.index) or not df.columns.equals(self.columns),
+                "Internal and external indices do not match.",
+            )
+            df.index = self.index
+            df.columns = self.columns
         return df
 
     def to_numpy(self):
diff --git a/modin/engines/ray/generic/io.py b/modin/engines/ray/generic/io.py
index f5723a4b633..e364903f1e0 100644
--- a/modin/engines/ray/generic/io.py
+++ b/modin/engines/ray/generic/io.py
@@ -41,6 +41,6 @@ def func(df):
             df.to_sql(**kwargs)
             return pandas.DataFrame()
 
-        result = qc._modin_frame._fold_reduce(1, func)
+        result = qc._modin_frame._apply_full_axis(1, func, new_index=[], new_columns=[])
         # blocking operation
         result.to_pandas()
diff --git a/modin/pandas/groupby.py b/modin/pandas/groupby.py
index d87c79e2d5f..cc9ba04d280 100644
--- a/modin/pandas/groupby.py
+++ b/modin/pandas/groupby.py
@@ -295,7 +295,7 @@ def __getitem__(self, key):
                 and all(c in self._columns for c in self._by.columns)
                 and self._drop
             ):
-                key = [key] + list(self._by.columns)
+                key = list(self._by.columns) + [key]
             else:
                 key = [key]
         if isinstance(key, list) and (make_dataframe or not self._as_index):
diff --git a/modin/pandas/test/test_general.py b/modin/pandas/test/test_general.py
index deda3512af2..95006d9d61e 100644
--- a/modin/pandas/test/test_general.py
+++ b/modin/pandas/test/test_general.py
@@ -412,3 +412,28 @@ def test_to_numeric(data, errors, downcast):
     modin_result = pd.to_numeric(modin_series, errors=errors, downcast=downcast)
     pandas_result = pandas.to_numeric(pandas_series, errors=errors, downcast=downcast)
     df_equals(modin_result, pandas_result)
+
+
+def test_to_pandas_indices():
+    data = test_data_values[0]
+
+    md_df = pd.DataFrame(data)
+    index = pandas.MultiIndex.from_tuples(
+        [(i, i * 2) for i in np.arange(len(md_df) + 1)], names=["A", "B"]
+    ).drop(0)
+    columns = pandas.MultiIndex.from_tuples(
+        [(i, i * 2) for i in np.arange(len(md_df.columns) + 1)], names=["A", "B"]
+    ).drop(0)
+
+    md_df.index = index
+    md_df.columns = columns
+
+    pd_df = md_df._to_pandas()
+
+    for axis in [0, 1]:
+        assert md_df.axes[axis].equals(
+            pd_df.axes[axis]
+        ), f"Indices at axis {axis} are different!"
+        assert md_df.axes[axis].equal_levels(
+            pd_df.axes[axis]
+        ), f"Levels of indices at axis {axis} are different!"
diff --git a/modin/pandas/test/test_io.py b/modin/pandas/test/test_io.py
index e63ff86c9af..0331f5013b3 100644
--- a/modin/pandas/test/test_io.py
+++ b/modin/pandas/test/test_io.py
@@ -559,6 +559,7 @@ def test_from_clipboard():
     df_equals(modin_df, pandas_df)
 
 
+@pytest.mark.xfail(reason="read_excel is broken for now, see #1733 for details")
 def test_from_excel():
     setup_excel_file(SMALL_ROW_SIZE)
 
diff --git a/modin/pandas/test/test_series.py b/modin/pandas/test/test_series.py
index 6eb3169f559..53d2764273f 100644
--- a/modin/pandas/test/test_series.py
+++ b/modin/pandas/test/test_series.py
@@ -1362,6 +1362,9 @@ def test_dtype(data):
     df_equals(modin_series.dtype, pandas_series.dtypes)
 
 
+@pytest.mark.xfail(
+    reason="Datetime properties is broken for now, see #1729 for details"
+)
 def test_dt():
     data = pd.date_range("2016-12-31", "2017-01-08", freq="D", tz="Europe/Berlin")
     modin_series = pd.Series(data)


Code review comments:
diff_hunk: @@ -234,6 +256,74 @@ def _filter_empties(self):
         self._column_widths_cache = [w for w in self._column_widths if w > 0]
         self._row_lengths_cache = [r for r in self._row_lengths if r > 0]
 
+    def _validate_axis_equality(self, axis: int, update: bool):
+        """
+        Validates internal and external indices of modin_frame at the specified axis.
+
+        Parameters
+        ----------
+            axis : int,
+                Axis to validate indices along
+            update : bool,
+                Update or raise an exception if indices don't match
+        """
+        internal_axis = self._frame_mgr_cls.get_indices(
+            axis, self._partitions, lambda df: df.axes[axis]
+        )
+        is_equals = self.axes[axis].equals(internal_axis)
+        if not update:
path: modin/engines/base/frame/data.py
text: @user1:
I did not see this getting used in this PR, how are you expecting this to be used?

@user1:
It may be better to just leave the `update` parameter out because I don't think we ever want to throw the internal error. What do you think?

@author:
well, if you think so

removed that parameter in the last commit