FIX-#5635: Correct reset_index column collision check for custom index names

modin/pandas/base.py

@@ -2689,19 +2689,73 @@ def reset_index(
         Reset the index, or a level of it.
         """
         inplace = validate_bool_kwarg(inplace, "inplace")
+        if allow_duplicates is not lib.no_default:
+            allow_duplicates = validate_bool_kwarg(allow_duplicates, "allow_duplicates")
         # Error checking for matching pandas. Pandas does not allow you to
-        # insert a dropped index into a DataFrame if these columns already
-        # exist.
+        # insert an index into DataFrame columns if these columns already exist.
         if (
             not drop
+            and (allow_duplicates is lib.no_default or not allow_duplicates)
             and not (
                 self._query_compiler.lazy_column_labels
                 or self._query_compiler.lazy_row_labels
             )
-            and not self._query_compiler.has_multiindex()
-            and all(n in self.columns for n in ["level_0", "index"])
         ):
-            raise ValueError("cannot insert level_0, already exists")
+            if level is not None:
+                if not isinstance(level, (tuple, list)):
+                    levels = [level]
+                else:
+                    levels = list(level)
+                levels = [self.index._get_level_number(lev) for lev in levels]
+            else:
+                levels = list(range(self.index.nlevels))
+
+            default_name = "index" if "index" not in self.columns else "level_0"
+            if hasattr(self.index, "_get_default_index_names"):
+                names_list = self.index._get_default_index_names(names, default_name)
+            elif names is not None:
+                if isinstance(names, (int, str)):
+                    names_list = [names]
+                elif not isinstance(names, list):
+                    names_list = list(names)
+                else:
+                    names_list = names
+            elif isinstance(self.index, pandas.MultiIndex):
+                names_list = [
+                    f"level_{i}" if n is None else n
+                    for i, n in enumerate(self.index.names)
+                ]
+            else:
+                names_list = (
+                    [default_name] if self.index.name is None else [self.index.name]
+                )
+
+            multi_col = isinstance(self.columns, pandas.MultiIndex)
+            new_cols = list(self.columns)
+
+            for i in reversed(range(self.index.nlevels)):
+                if i not in levels:
+                    continue
+                name = names_list[i]
+                if multi_col:
+                    col_name = list(name) if isinstance(name, tuple) else [name]
+                    fill = col_fill
+                    if fill is None:
+                        if len(col_name) not in (1, self.columns.nlevels):
+                            raise ValueError(
+                                "col_fill=None is incompatible "
+                                f"with incomplete column name {name}"
+                            )
+                        fill = col_name[0]
+                    lev_num = self.columns._get_level_number(col_level)
+                    name_lst = [fill] * lev_num + col_name
+                    missing = self.columns.nlevels - len(name_lst)
+                    name_lst += [fill] * missing
+                    name = tuple(name_lst)
+
+                if name in new_cols:
+                    raise ValueError(f"cannot insert {name}, already exists")
+                new_cols.insert(0, name)
         new_query_compiler = self._query_compiler.reset_index(
             drop=drop,
             level=level,

modin/tests/pandas/dataframe/test_indexing.py

@@ -1835,6 +1835,43 @@ def test_reset_index_metadata_update(index, test_async_reset_index):
     eval_general(modin_df, pandas_df, lambda df: df.reset_index())
 
 
+@pytest.mark.parametrize("test_async_reset_index", [False, True])
+def test_reset_index_custom_name_collision(test_async_reset_index):
+    # GH#5635: reset_index should not raise ValueError when index has custom name
+    # even if columns contain 'level_0' and 'index'.
+    data = {"index": [1, 2], "level_0": [3, 4]}
+    index = pandas.Index([10, 20], name="custom")
+    modin_df, pandas_df = create_test_dfs(data, index=index)
+    if test_async_reset_index:
+        modin_df._query_compiler.set_frame_index_cache(None)
+    eval_general(modin_df, pandas_df, lambda df: df.reset_index())
+
+    # Should raise ValueError when custom name collides with existing column
+    modin_df_col, pandas_df_col = create_test_dfs({"custom": [1, 2]}, index=index)
+    eval_general(
+        modin_df_col,
+        pandas_df_col,
+        lambda df: df.reset_index(),
+        expected_exception=ValueError("cannot insert custom, already exists"),
+    )
+
+    # MultiIndex level name collision
+    midx = pandas.MultiIndex.from_tuples([(1, 2), (3, 4)], names=["a", "b"])
+    modin_midx, pandas_midx = create_test_dfs({"a": [10, 20]}, index=midx)
+    eval_general(
+        modin_midx,
+        pandas_midx,
+        lambda df: df.reset_index(),
+        expected_exception=ValueError("cannot insert a, already exists"),
+    )
+
+    # allow_duplicates=True allows duplicate columns
+    modin_dup, pandas_dup = create_test_dfs({"custom": [1, 2]}, index=index)
+    eval_general(
+        modin_dup, pandas_dup, lambda df: df.reset_index(allow_duplicates=True)
+    )
+
+
 @pytest.mark.parametrize("data", test_data_values, ids=test_data_keys)
 @pytest.mark.parametrize("axis", axis_values, ids=axis_keys)
 def test_sample(data, axis):

modin/tests/pandas/test_series.py

@@ -3204,6 +3204,24 @@ def test_reset_index(data, drop, name, inplace):
     )
 
 
+def test_reset_index_name_collision():
+    # GH#5635: Series.reset_index should raise ValueError when index name collides with series name
+    index = pandas.Index([10, 20], name="custom")
+    modin_s = pd.Series([1, 2], index=index, name="custom")
+    pandas_s = pandas.Series([1, 2], index=index, name="custom")
+    eval_general(
+        modin_s,
+        pandas_s,
+        lambda s: s.reset_index(),
+        expected_exception=ValueError("cannot insert custom, already exists"),
+    )
+
+    # Should succeed without collision
+    modin_s2 = pd.Series([1, 2], index=index, name="other")
+    pandas_s2 = pandas.Series([1, 2], index=index, name="other")
+    eval_general(modin_s2, pandas_s2, lambda s: s.reset_index())
+
+
 @pytest.mark.parametrize("data", test_data_values, ids=test_data_keys)
 def test_rfloordiv(data):
     modin_series, pandas_series = create_test_series(data)