Repository: dask/dask
File: dask/dataframe/dask_expr/_merge.py (Python)

Code before:
        # TODO:
        #   - Need 'rearrange_by_divisions' equivalent
        #     to avoid shuffle when we are merging on known
        #     divisions on one side only.
        else:
            if left_index:
                shuffle_left_on = left.index._meta.name
                if shuffle_left_on is None:
                    # placeholder for unnamed index merge
                    shuffle_left_on = "_index"
            if right_index:
                shuffle_right_on = right.index._meta.name
                if shuffle_right_on is None:
                    shuffle_right_on = "_index"
            if self.is_broadcast_join:
                left, right = self._bcast_left, self._bcast_right

                if self.how != "inner":
                    if self.broadcast_side == "left":
                        left = RearrangeByColumn(
                            left,
                            shuffle_left_on,
                            npartitions_out=left.npartitions,
                        )
                    else:
                        right = RearrangeByColumn(
                            right,
                            shuffle_right_on,
                            npartitions_out=right.npartitions,
                        )

                return BroadcastJoin(
                    left,
                    right,
                    self.how,
                    left_on,
                    right_on,
                    left_index,
                    right_index,
                    self.suffixes,
                    self.indicator,
                )

        if (shuffle_left_on or shuffle_right_on) and (
            shuffle_method == "p2p"
            or shuffle_method is None
            and get_default_shuffle_method() == "p2p"
            and not left_already_partitioned
            and not right_already_partitioned
        ):
            return HashJoinP2P(

Review comment (suggestion):
these shuffle already partitioned conditions are the only logic that changed slightly. I think everything else on this PR was just moving the various pieces of the base class _divisions to the subclasses this was actually relevant for, i.e. mostly BlockwiseMerge

Code after:
                how=self.how,
                left_on=left_on,
                right_on=right_on,
                suffixes=self.suffixes,
                indicator=self.indicator,
                left_index=left_index,
                right_index=right_index,
                shuffle_left_on=shuffle_left_on,
                shuffle_right_on=shuffle_right_on,
                _npartitions=shuffle_npartitions,
            )
        if shuffle_left_on and not (
            left_already_partitioned and self.left.npartitions == shuffle_npartitions
        ):
            # Shuffle left
            left = RearrangeByColumn(
                left,
                shuffle_left_on,
                npartitions_out=shuffle_npartitions,
                method=shuffle_method,
                index_shuffle=left_index,
            )

        if shuffle_right_on and not (
            right_already_partitioned and self.right.npartitions == shuffle_npartitions
        ):
            # Shuffle right
            right = RearrangeByColumn(
                right,
                shuffle_right_on,
                npartitions_out=shuffle_npartitions,
                method=shuffle_method,
                index_shuffle=right_index,
            )

        # Blockwise merge
        return BlockwiseMerge(left, right, **self.kwargs)

    def _simplify_up(self, parent, dependents):
        if isinstance(parent, Filter):
            if not self._filter_passthrough_available(parent, dependents):
                return
            predicate = parent.predicate

            if isinstance(predicate, And):
                new = Filter(self, predicate.left)
                new_pred = predicate.right.substitute(self, new)
                return Filter(new, new_pred)

            predicate_cols = self._predicate_columns(parent.predicate)
            new_left, new_right = self.left, self.right