| 723 | |
| 724 | @derived_from(pd.DataFrame) |
| 725 | def isin(self, values): |
| 726 | if isinstance(self, DataFrame): |
| 727 | # DataFrame.isin does weird alignment stuff |
| 728 | bad_types = (FrameBase, pd.Series, pd.DataFrame) |
| 729 | else: |
| 730 | bad_types = (FrameBase,) |
| 731 | if isinstance(values, bad_types): |
| 732 | if ( |
| 733 | isinstance(values, FrameBase) |
| 734 | and values.ndim == 1 |
| 735 | and values.npartitions == 1 |
| 736 | ): |
| 737 | # Can broadcast |
| 738 | return new_collection(expr.Isin(self, values=values)) |
| 739 | raise NotImplementedError("Passing a %r to `isin`" % typename(type(values))) |
| 740 | |
| 741 | # We wrap values in a delayed for two reasons: |
| 742 | # - avoid serializing data in every task |
| 743 | # - avoid cost of traversal of large list in optimizations |
| 744 | if isinstance(values, list): |
| 745 | # Motivated by https://github.com/dask/dask/issues/9411. This appears to be |
| 746 | # caused by https://github.com/dask/distributed/issues/6368, and further |
| 747 | # exacerbated by the fact that the list contains duplicates. This is a patch until |
| 748 | # we can create a better fix for Serialization. |
| 749 | try: |
| 750 | values = list(set(values)) |
| 751 | except TypeError: |
| 752 | pass |
| 753 | if not any(is_dask_collection(v) for v in values): |
| 754 | # Avoid always passing a numpy array of object dtype |
| 755 | inferred_type = pd.api.types.infer_dtype(values, skipna=False) |
| 756 | object_like = { |
| 757 | "mixed-integer", |
| 758 | "decimal", |
| 759 | "categorical", |
| 760 | "time", |
| 761 | "period", |
| 762 | "mixed", |
| 763 | "unknown-array", |
| 764 | } |
| 765 | if inferred_type in object_like: |
| 766 | values = np.fromiter(values, dtype=object, count=len(values)) |
| 767 | else: |
| 768 | values = np.asarray(values) |
| 769 | |
| 770 | return new_collection( |
| 771 | expr.Isin( |
| 772 | self, |
| 773 | values=expr._DelayedExpr( |
| 774 | delayed(values, name="delayed-" + _tokenize_deterministic(values)) |
| 775 | ), |
| 776 | ) |
| 777 | ) |
| 778 | |
| 779 | def _partitions(self, index): |
| 780 | # Used by `partitions` for partition-wise slicing |