Match pandas merge/join dtype, index and error semantics

python/cudf/cudf/core/dataframe.py

@@ -4993,6 +4993,24 @@ def merge(
             left_on, right_on = right_on, left_on
             left_index, right_index = right_index, left_index
             suffixes = (suffixes[1], suffixes[0])
+            if (
+                orig_on is None
+                and orig_left_on is None
+                and orig_right_on is None
+                and not orig_left_index
+                and not orig_right_index
+            ):
+                # Merge infers the common key columns from its (post-swap)
+                # left frame, but pandas keys an inferred merge by the
+                # *original* left frame's column order regardless of ``how``.
+                # Pass the keys explicitly to preserve that order (it decides
+                # both the key column order and the sort priority).
+                right_names = set(right._column_names)
+                inferred_on = [
+                    name for name in self._column_names if name in right_names
+                ]
+                if inferred_on:
+                    on = inferred_on
         elif how in {"leftsemi", "leftanti"}:
             merge_cls = MergeSemi
 
@@ -5051,8 +5069,28 @@ def merge(
                 and not orig_left_index
                 and not orig_right_index
             ):
-                # Auto-detect: intersection of column names are the keys.
-                k = len(set(self._column_names) & set(right._column_names))
+                # Auto-detected keys sit wherever they appear within each
+                # frame, so a segment swap cannot reproduce pandas' layout
+                # (the original left frame's columns in their own order,
+                # then the right frame's non-key columns). No suffixing can
+                # occur here -- any label shared by both frames is a key --
+                # so the result labels are unchanged and a label-based
+                # reorder is exact.
+                common = set(self._column_names) & set(right._column_names)
+                expected = list(self._column_names) + [
+                    name for name in right._column_names if name not in common
+                ]
+                positions = {
+                    label: i for i, label in enumerate(result._column_names)
+                }
+                if len(expected) == n_result and all(
+                    label in positions for label in expected
+                ):
+                    result = result.iloc[
+                        :, [positions[label] for label in expected]
+                    ]
+                # skip the positional segment swap below
+                k = n_result
             else:
                 k = 0
             # Only reorder when there are both right non-key cols and self
@@ -5065,6 +5103,98 @@ def merge(
                 )
                 result = result.iloc[:, new_indices]
 
+        if how != "cross":
+            orig_how = "right" if is_right_join else how
+            if is_right_join:
+                key_on, key_lon, key_ron = orig_on, orig_left_on, orig_right_on
+                key_li, key_ri = orig_left_index, orig_right_index
+            else:
+                key_on, key_lon, key_ron = on, left_on, right_on
+                key_li, key_ri = left_index, right_index
+
+            def _restore_key_dtype(name, target, right_dtype):
+                # Restore a result key column to ``target`` to match pandas.
+                if name not in result._data:
+                    return
+                # Categorical keys follow the (de)categorization rules applied
+                # during the join itself.
+                if isinstance(target, CategoricalDtype) or isinstance(
+                    right_dtype, CategoricalDtype
+                ):
+                    return
+                if result._data[name].dtype == target:
+                    return
+                # Do not undo the numpy int -> float64 upcast that unmatched
+                # rows require.
+                if (
+                    isinstance(target, np.dtype)
+                    and result._data[name].null_count
+                ):
+                    return
+                result[name] = result[name].astype(target)
+
+            def _keep_left_dtype(target, right_dtype):
+                # pandas presents a shared-name key with the LEFT dtype when an
+                # extension dtype is involved (all joins) or for inner/left
+                # joins; right/outer numpy keys take the common type.
+                one_extension = (not isinstance(target, np.dtype)) or (
+                    right_dtype is not None
+                    and not isinstance(right_dtype, np.dtype)
+                )
+                return one_extension or orig_how in {
+                    "inner",
+                    "left",
+                    "leftsemi",
+                    "leftanti",
+                }
+
+            if not key_li and not key_ri:
+                if key_lon is not None and key_ron is not None:
+                    lon = [key_lon] if is_scalar(key_lon) else list(key_lon)
+                    ron = [key_ron] if is_scalar(key_ron) else list(key_ron)
+                    for lk, rk in zip(lon, ron, strict=True):
+                        if lk == rk:
+                            if lk in self._data:
+                                target = self._data[lk].dtype
+                                rd = (
+                                    right._data[lk].dtype
+                                    if lk in right._data
+                                    else None
+                                )
+                                if _keep_left_dtype(target, rd):
+                                    _restore_key_dtype(lk, target, rd)
+                        else:
+                            # Differently-named keys both survive, each with
+                            # its own operand's dtype.
+                            if lk in self._data:
+                                _restore_key_dtype(
+                                    lk, self._data[lk].dtype, None
+                                )
+                            if rk in right._data:
+                                _restore_key_dtype(
+                                    rk, right._data[rk].dtype, None
+                                )
+                else:
+                    if key_on is not None:
+                        key_names = (
+                            [key_on] if is_scalar(key_on) else list(key_on)
+                        )
+                    else:
+                        key_names = list(
+                            set(self._column_names) & set(right._column_names)
+                        )
+                    for name in key_names:
+                        if name not in self._data:
+                            continue
+                        target = self._data[name].dtype
+                        rd = (
+                            right._data[name].dtype
+                            if name in right._data
+                            else None
+                        )
+                        if _keep_left_dtype(target, rd):
+                            _restore_key_dtype(name, target, rd)
+
         return result
 
     @_performance_tracking

python/cudf/cudf/core/index.py

@@ -848,10 +848,25 @@ def union(self, other, sort: bool | None = None) -> Index:
         return result
 
     def _intersection(self, other, sort: bool | None = None) -> Index:
+        lcol = self.unique()._column
+        rcol = other.unique()._column
+        if (
+            is_dtype_obj_numeric(self.dtype, include_decimal=False)
+            and is_dtype_obj_numeric(other.dtype, include_decimal=False)
+            and self.dtype != other.dtype
+        ):
+            # pandas casts mismatched numeric dtypes to their common type
+            # for set operations ("cast to float, not object"); cast up
+            # front so the merge's key-dtype rules (which keep the left
+            # operand's dtype for inner joins) don't leak the left dtype
+            # into the result.
+            common_dtype = find_common_type([self.dtype, other.dtype])
+            lcol = lcol.astype(common_dtype)
+            rcol = rcol.astype(common_dtype)
         intersection_result = _index_from_data(
-            cudf.DataFrame._from_data({"None": self.unique()._column})
+            cudf.DataFrame._from_data({"None": lcol})
             .merge(
-                cudf.DataFrame._from_data({"None": other.unique()._column}),
+                cudf.DataFrame._from_data({"None": rcol}),
                 how="inner",
                 on="None",
             )

python/cudf/cudf/core/indexed_frame.py

@@ -211,7 +211,55 @@ def _indices_from_labels(obj, labels):
     rhs = cudf.DataFrame(
         {"_": ColumnBase.from_range(range(len(obj)))}, index=obj.index
     )
-    return lhs.join(rhs).sort_values(by=["__", "_"])["_"]
+    result = lhs.join(rhs).sort_values(by=["__", "_"])["_"]
+    if result.dtype != rhs._data["_"].dtype:
+        # The merge upcasts the positional column to float64 when some
+        # labels are missing; row positions are exactly representable in
+        # float64, so restore the integer dtype (nulls are preserved).
+        result = result.astype(rhs._data["_"].dtype)
+    return result
+
+
+def _gather_map_from_positions(positions, nrows: int) -> GatherMap:
+    """Build a nullifying GatherMap from a joined positional column.
+
+    A merge that leaves some rows unmatched upcasts a positional column to
+    float64 with nulls; row positions are exactly representable in float64,
+    so cast back and replace nulls with an out-of-bounds sentinel that a
+    nullifying gather turns back into nulls.
+    """
+    positions = positions.astype(SIZE_TYPE_DTYPE)
+    if positions.null_count:
+        positions = positions.fillna(np.int32(np.iinfo(np.int32).min))
+    return GatherMap.from_column_unchecked(positions, nrows, nullify=True)
+
+
+def _unify_categorical_indexes(lhs_index, rhs_index):
+    """Cast two unordered categorical indexes with differing categories to
+    a common merged-categories dtype ahead of a join.
+
+    The merged categories keep first-appearance order (the left categories
+    followed by the right's unseen ones), so a sorted join on the unified
+    codes orders rows by category -- like pandas' union of categoricals --
+    rather than lexically, and the joined index remains categorical.
+    """
+    ldtype = getattr(lhs_index, "dtype", None)
+    rdtype = getattr(rhs_index, "dtype", None)
+    if (
+        isinstance(ldtype, cudf.CategoricalDtype)
+        and isinstance(rdtype, cudf.CategoricalDtype)
+        and not ldtype.ordered
+        and not rdtype.ordered
+        and not ldtype._internal_eq(rdtype)
+    ):
+        merged_categories = cudf.concat(
+            [ldtype.categories, rdtype.categories]
+        ).unique()
+        common = cudf.CategoricalDtype(
+            categories=merged_categories, ordered=False
+        )
+        return lhs_index.astype(common), rhs_index.astype(common)
+    return lhs_index, rhs_index
 
 
 class _FrameIndexer:
@@ -3882,24 +3930,35 @@ def _align_to_index(
             if not self.index.is_unique or not index.is_unique:
                 raise ValueError("Cannot align indices with non-unique values")
 
-        lhs = cudf.DataFrame._from_data(self._data, index=self.index)
-        rhs = cudf.DataFrame._from_data({}, index=index)
+        # Join only the indexes, with a positional column standing in for
+        # this frame's rows, and gather the data columns afterwards. Routing
+        # the data columns through the merge would subject them to pandas'
+        # merge dtype semantics (e.g. the numpy int -> float64 upcast on
+        # unmatched rows), which apply to user-facing merges but not to
+        # alignment.
+        lhs_index, rhs_index = _unify_categorical_indexes(self.index, index)
+        pos_col_id = str(uuid4())
+        lhs = cudf.DataFrame._from_data(
+            {pos_col_id: ColumnBase.from_range(range(len(self)))},
+            index=lhs_index,
+        )
+        rhs = cudf.DataFrame._from_data({}, index=rhs_index)
 
         # create a temporary column that we will later sort by
         # to recover ordering after index alignment.
         sort_col_id = str(uuid4())
-        if how == "left":
-            lhs[sort_col_id] = ColumnBase.from_range(range(len(lhs)))
-        elif how == "right":
+        if how == "right":
             rhs[sort_col_id] = ColumnBase.from_range(range(len(rhs)))
 
         result = lhs.join(rhs, how=how, sort=sort)
-        if how in ("left", "right"):
+        if how == "left":
+            result = result.sort_values(pos_col_id)
+        elif how == "right":
             result = result.sort_values(sort_col_id)
-            del result[sort_col_id]
 
-        out = self._from_data(
-            self._data._from_columns_like_self(result._columns)
+        out = self._gather(
+            _gather_map_from_positions(result._data[pos_col_id], len(self)),
+            keep_index=False,
         )
         out.index = result.index
         out.index.names = self.index.names
@@ -3995,7 +4054,6 @@ def _reindex(
                 )
                 df = cudf.DataFrame()
             else:
-                lhs = cudf.DataFrame._from_data({}, index=index)
                 rhs = cudf.DataFrame._from_data(
                     {
                         # bookkeeping workaround for unnamed series
@@ -4008,11 +4066,36 @@ def _reindex(
                 )
                 diff = index.difference(df.index)
                 rows_added = len(diff) > 0
-                df = lhs.join(rhs, how="left", sort=True)
+                # Join only the indexes and gather the data columns natively
+                # (see ``_align_to_index``): the merge's pandas dtype
+                # semantics must not leak into reindexing, whose own dtype
+                # rules are applied below. Row order comes from a positional
+                # column on the target rather than from sorting the joined
+                # key: a value sort's collation can disagree with the
+                # target's own ordering (e.g. a categorical target joined
+                # against a string source is decategorized and would sort
+                # lexically rather than by category).
+                pos_col_id = str(uuid4())
+                order_col_id = str(uuid4())
+                lhs = cudf.DataFrame._from_data(
+                    {order_col_id: ColumnBase.from_range(range(len(index)))},
+                    index=index,
+                )
+                pos_rhs = cudf.DataFrame._from_data(
+                    {pos_col_id: ColumnBase.from_range(range(len(rhs)))},
+                    index=rhs.index,
+                )
+                joined = lhs.join(pos_rhs, how="left", sort=False)
+                joined = joined.sort_values(order_col_id)
+                df = rhs._gather(
+                    _gather_map_from_positions(
+                        joined._data[pos_col_id], len(rhs)
+                    ),
+                    keep_index=False,
+                )
+                df.index = joined.index
                 if fill_value is not NA and rows_added:
                     df.loc[diff] = fill_value
-                # double-argsort to map back from sorted to unsorted positions
-                df = df.take(index.argsort(ascending=True).argsort())
 
         index = index if index is not None else df.index