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

Code before:
            self.clear_divisions,
            self.align_dataframes,
            None,
            None,
            self.token,
            self._kwargs,
            len(self.args[1:]),
            *self.args[1:],
        )


class CreateOverlappingPartitions(Expr):
    _parameters = ["frame", "before", "after"]

    @functools.cached_property
    def _meta(self):
        return self.frame._meta

    def _divisions(self):
        # Keep divisions alive, MapPartitions will handle the actual division logic
        return self.frame.divisions

    def _layer(self) -> dict:
        dsk, prevs, nexts = {}, [], []  # type: ignore

        name_prepend = "overlap-prepend" + tokenize(self.frame._name, self.before)
        if self.before:
            prevs.append(None)
            if isinstance(self.before, numbers.Integral):
                before = self.before
                for i in range(self.frame.npartitions - 1):
                    dsk[(name_prepend, i)] = (M.tail, (self.frame._name, i), before)
                    prevs.append((name_prepend, i))
            elif isinstance(self.before, datetime.timedelta):
                # Assumes monotonic (increasing?) index
                divs = pd.Series(self.frame.divisions)
                deltas = divs.diff().iloc[1:-1]

                # In the first case window-size is larger than at least one partition, thus it is
                # necessary to calculate how many partitions must be used for each rolling task.
                # Otherwise, these calculations can be skipped (faster)

                if (self.before > deltas).any():
                    pt_z = divs[0]
                    for i in range(self.frame.npartitions - 1):
                        # Select all indexes of relevant partitions between the current partition and
                        # the partition with the highest division outside the rolling window (before)
                        pt_i = divs[i + 1]

                        # lower-bound the search to the first division
                        lb = max(pt_i - self.before, pt_z)

Review comment (suggestion):
```suggestion
        name_prepend = "overlap-prepend-" + self._name
```
Pretty sure we always want the name to be unique to this specific `Expr` object (not its parent). Not sure why this was `self.frame._name` before.

Code after:
            self.clear_divisions,
            self.align_dataframes,
            None,
            None,
            self.token,
            self._kwargs,
            len(self.args[1:]),
            *self.args[1:],
        )


class CreateOverlappingPartitions(Expr):
    _parameters = ["frame", "before", "after"]

    @functools.cached_property
    def _meta(self):
        return self.frame._meta

    def _divisions(self):
        # Keep divisions alive, MapPartitions will handle the actual division logic
        return self.frame.divisions

    def _layer(self) -> dict:
        dsk, prevs, nexts = {}, [], []  # type: ignore

        name_prepend = "overlap-prepend-" + self._name
        if self.before:
            prevs.append(None)
            if isinstance(self.before, numbers.Integral):
                before = self.before
                for i in range(self.frame.npartitions - 1):
                    dsk[(name_prepend, i)] = (M.tail, (self.frame._name, i), before)
                    prevs.append((name_prepend, i))
            elif isinstance(self.before, datetime.timedelta):
                # Assumes monotonic (increasing?) index
                divs = pd.Series(self.frame.divisions)
                deltas = divs.diff().iloc[1:-1]

                # In the first case window-size is larger than at least one partition, thus it is
                # necessary to calculate how many partitions must be used for each rolling task.
                # Otherwise, these calculations can be skipped (faster)

                if (self.before > deltas).any():
                    pt_z = divs[0]
                    for i in range(self.frame.npartitions - 1):
                        # Select all indexes of relevant partitions between the current partition and
                        # the partition with the highest division outside the rolling window (before)
                        pt_i = divs[i + 1]

                        # lower-bound the search to the first division
                        lb = max(pt_i - self.before, pt_z)