ENH: Implement PeriodIndex.intersection without object-dtype cast (#30666)

Author: jbrockmendel

pandas/core/indexes/base.py

@@ -2408,14 +2408,8 @@ def intersection(self, other, sort=False):
             return this.intersection(other, sort=sort)
 
         # TODO(EA): setops-refactor, clean all this up
-        if is_period_dtype(self):
-            lvals = self._ndarray_values
-        else:
-            lvals = self._values
-        if is_period_dtype(other):
-            rvals = other._ndarray_values
-        else:
-            rvals = other._values
+        lvals = self._values
+        rvals = other._values
 
         if self.is_monotonic and other.is_monotonic:
             try:
@@ -2434,18 +2428,13 @@ def intersection(self, other, sort=False):
             indexer = indexer[indexer != -1]
 
         taken = other.take(indexer)
+        res_name = get_op_result_name(self, other)
 
         if sort is None:
             taken = algos.safe_sort(taken.values)
-            if self.name != other.name:
-                name = None
-            else:
-                name = self.name
-            return self._shallow_copy(taken, name=name)
-
-        if self.name != other.name:
-            taken.name = None
+            return self._shallow_copy(taken, name=res_name)
 
+        taken.name = res_name
         return taken
 
     def difference(self, other, sort=None):

pandas/core/indexes/datetimelike.py

@@ -721,12 +721,9 @@ def intersection(self, other, sort=False):
             # this point, depending on the values.
 
             result._set_freq(None)
-            if hasattr(self, "tz"):
-                result = self._shallow_copy(
-                    result._values, name=result.name, tz=result.tz, freq=None
-                )
-            else:
-                result = self._shallow_copy(result._values, name=result.name, freq=None)
+            result = self._shallow_copy(
+                result._data, name=result.name, dtype=result.dtype, freq=None
+            )
             if result.freq is None:
                 result._set_freq("infer")
             return result

pandas/core/indexes/period.py

@@ -12,6 +12,7 @@
     ensure_platform_int,
     is_bool_dtype,
     is_datetime64_any_dtype,
+    is_dtype_equal,
     is_float,
     is_float_dtype,
     is_integer,
@@ -782,6 +783,9 @@ def join(self, other, how="left", level=None, return_indexers=False, sort=False)
             return self._apply_meta(result), lidx, ridx
         return self._apply_meta(result)
 
+    # ------------------------------------------------------------------------
+    # Set Operation Methods
+
     def _assert_can_do_setop(self, other):
         super()._assert_can_do_setop(other)
 
@@ -799,6 +803,30 @@ def _wrap_setop_result(self, other, result):
         result.name = name
         return result
 
+    def intersection(self, other, sort=False):
+        self._validate_sort_keyword(sort)
+        self._assert_can_do_setop(other)
+        res_name = get_op_result_name(self, other)
+        other = ensure_index(other)
+
+        if self.equals(other):
+            return self._get_reconciled_name_object(other)
+
+        if not is_dtype_equal(self.dtype, other.dtype):
+            # TODO: fastpath for if we have a different PeriodDtype
+            this = self.astype("O")
+            other = other.astype("O")
+            return this.intersection(other, sort=sort)
+
+        i8self = Int64Index._simple_new(self.asi8)
+        i8other = Int64Index._simple_new(other.asi8)
+        i8result = i8self.intersection(i8other, sort=sort)
+
+        result = self._shallow_copy(np.asarray(i8result, dtype=np.int64), name=res_name)
+        return result
+
+    # ------------------------------------------------------------------------
+
     def _apply_meta(self, rawarr):
         if not isinstance(rawarr, PeriodIndex):
             rawarr = PeriodIndex._simple_new(rawarr, freq=self.freq, name=self.name)