refactor: Respect session default index in merge and reset_index methods (#862)

Author: TrevorBergeron

bigframes/core/blocks.py

@@ -327,18 +327,30 @@ def reset_index(self, drop: bool = True) -> Block:
             A new Block because dropping index columns can break references
             from Index classes that point to this block.
         """
-        new_index_col_id = guid.generate_guid()
-        expr = self._expr.promote_offsets(new_index_col_id)
+        expr = self._expr
+        if (
+            self.session._default_index_type
+            == bigframes.enums.DefaultIndexKind.SEQUENTIAL_INT64
+        ):
+            new_index_col_id = guid.generate_guid()
+            expr = expr.promote_offsets(new_index_col_id)
+            new_index_cols = [new_index_col_id]
+        elif self.session._default_index_type == bigframes.enums.DefaultIndexKind.NULL:
+            new_index_cols = []
+        else:
+            raise ValueError(
+                f"Unrecognized default index kind: {self.session._default_index_type}"
+            )
+
         if drop:
             # Even though the index might be part of the ordering, keep that
             # ordering expression as reset_index shouldn't change the row
             # order.
             expr = expr.drop_columns(self.index_columns)
             return Block(
                 expr,
-                index_columns=[new_index_col_id],
+                index_columns=new_index_cols,
                 column_labels=self.column_labels,
-                index_labels=[None],
             )
         else:
             # Add index names to column index
@@ -362,9 +374,8 @@ def reset_index(self, drop: bool = True) -> Block:
 
             return Block(
                 expr,
-                index_columns=[new_index_col_id],
+                index_columns=new_index_cols,
                 column_labels=column_labels_modified,
-                index_labels=[None],
             )
 
     def set_index(
@@ -2096,13 +2107,17 @@ def merge(
         #
         # This keeps us from generating an index if the user joins a large
         # BigQuery table against small local data, for example.
-        if len(self._index_columns) > 0 and len(other._index_columns) > 0:
+        if (
+            self.index.is_null
+            or other.index.is_null
+            or self.session._default_index_type == bigframes.enums.DefaultIndexKind.NULL
+        ):
+            expr = joined_expr
+            index_columns = []
+        else:
             offset_index_id = guid.generate_guid()
             expr = joined_expr.promote_offsets(offset_index_id)
             index_columns = [offset_index_id]
-        else:
-            expr = joined_expr
-            index_columns = []
 
         return Block(expr, index_columns=index_columns, column_labels=labels)
 
@@ -2604,6 +2619,10 @@ def column_ids(self) -> Sequence[str]:
         """Column(s) to use as row labels."""
         return self._block._index_columns
 
+    @property
+    def is_null(self) -> bool:
+        return len(self._block._index_columns) == 0
+
     def to_pandas(self, *, ordered: Optional[bool] = None) -> pd.Index:
         """Executes deferred operations and downloads the results."""
         if len(self.column_ids) == 0:

tests/system/small/test_unordered.py

@@ -116,6 +116,28 @@ def test_unordered_drop_duplicates(unordered_session, keep):
     assert_pandas_df_equal(bf_result.to_pandas(), pd_result, ignore_order=True)
 
 
+def test_unordered_reset_index(unordered_session):
+    pd_df = pd.DataFrame({"a": [1, 1, 3], "b": [4, 4, 6]}, dtype=pd.Int64Dtype())
+    bf_df = bpd.DataFrame(pd_df, session=unordered_session)
+
+    bf_result = bf_df.set_index("b").reset_index(drop=False)
+    pd_result = pd_df.set_index("b").reset_index(drop=False)
+
+    assert_pandas_df_equal(bf_result.to_pandas(), pd_result)
+
+
+def test_unordered_merge(unordered_session):
+    pd_df = pd.DataFrame(
+        {"a": [1, 1, 3], "b": [4, 4, 6], "c": [1, 2, 3]}, dtype=pd.Int64Dtype()
+    )
+    bf_df = bpd.DataFrame(pd_df, session=unordered_session)
+
+    bf_result = bf_df.merge(bf_df, left_on="a", right_on="c")
+    pd_result = pd_df.merge(pd_df, left_on="a", right_on="c")
+
+    assert_pandas_df_equal(bf_result.to_pandas(), pd_result, ignore_order=True)
+
+
 @pytest.mark.parametrize(
     ("function"),
     [