REF: de-duplicate Index._intersection + MultiIndex._intersection (pandas-dev#41824)

Author: jbrockmendel

pandas/core/indexes/base.py

@@ -3098,7 +3098,6 @@ def _intersection(self, other: Index, sort=False):
         intersection specialized to the case with matching dtypes.
         """
         # TODO(EA): setops-refactor, clean all this up
-        lvals = self._values
 
         if self.is_monotonic and other.is_monotonic:
             try:
@@ -3110,21 +3109,35 @@ def _intersection(self, other: Index, sort=False):
                 res = algos.unique1d(result)
                 return ensure_wrapped_if_datetimelike(res)
 
-        try:
-            indexer = other.get_indexer(lvals)
-        except InvalidIndexError:
-            # InvalidIndexError raised by get_indexer if non-unique
-            indexer, _ = other.get_indexer_non_unique(lvals)
+        res_values = self._intersection_via_get_indexer(other, sort=sort)
+        res_values = _maybe_try_sort(res_values, sort)
+        return res_values
 
-        mask = indexer != -1
-        indexer = indexer.take(mask.nonzero()[0])
+    def _intersection_via_get_indexer(self, other: Index, sort) -> ArrayLike:
+        """
+        Find the intersection of two Indexes using get_indexer.
+
+        Returns
+        -------
+        np.ndarray or ExtensionArray
+            The returned array will be unique.
+        """
+        # Note: drop_duplicates vs unique matters for MultiIndex, though
+        #  it should not, see GH#41823
+        left_unique = self.drop_duplicates()
+        right_unique = other.drop_duplicates()
 
-        result = other.take(indexer).unique()._values
-        result = _maybe_try_sort(result, sort)
+        indexer = left_unique.get_indexer(right_unique)
+
+        mask = indexer != -1
 
-        # Intersection has to be unique
-        assert Index(result).is_unique
+        taker = indexer.take(mask.nonzero()[0])
+        if sort is False:
+            # sort bc we want the elements in the same order they are in self
+            # unnecessary in the case with sort=None bc we will sort later
+            taker = np.sort(taker)
 
+        result = left_unique.take(taker)._values
         return result
 
     @final

pandas/core/indexes/multi.py

@@ -3612,9 +3612,7 @@ def _intersection(self, other, sort=False) -> MultiIndex:
                 uniq_tuples = algos.unique(inner_tuples)
 
         if uniq_tuples is None:
-            left_unique = self.drop_duplicates()
-            indexer = left_unique.get_indexer(other.drop_duplicates())
-            uniq_tuples = left_unique.take(np.sort(indexer[indexer != -1]))
+            uniq_tuples = self._intersection_via_get_indexer(other, sort)
 
         if sort is None:
             uniq_tuples = sorted(uniq_tuples)