Method _lower

dask/dataframe/dask_expr/_reductions.py:172–285

(self)

            return 1

    def _lower(self):
        from dask.dataframe.dask_expr._repartition import Repartition
        from dask.dataframe.dask_expr._shuffle import (
            RearrangeByColumn,
            SetIndexBlockwise,
            SortValues,
        )

        if is_index_like(self.frame._meta):
            columns = [
                (
                    self.frame._meta.name
                    if self.frame._meta.name is not None
                    else "__index__"
                )
            ]
        elif is_series_like(self.frame._meta):
            columns = [
                (
                    self.frame._meta.name
                    if self.frame._meta.name is not None
                    else "__series__"
                )
            ]
        else:
            columns = self.frame.columns

        # Find what columns we are shuffling by
        split_by = self.split_by or columns
        if not isinstance(split_by, (list, tuple)):
            split_by = [split_by]
        split_by_index = bool(set(split_by) - set(columns))

        # Make sure we have dataframe-like data to shuffle
        if split_by_index:
            if self.frame.ndim == 1:
                chunked = ResetIndex(self.frame, drop=False, name=self.frame.name)
            else:
                chunked = ResetIndex(self.frame, drop=False)
            if split_by == [None]:
                split_by = ["index"]
        elif is_index_like(self.frame._meta) or is_series_like(self.frame._meta):
            chunked = ToFrame(self.frame, name=columns[0])
        else:
            chunked = self.frame

        # Map Tuple[str] column names to str before the shuffle
        map_columns = {col: str(col) for col in chunked.columns if col != str(col)}
        unmap_columns = {v: k for k, v in map_columns.items()}
        if map_columns:
            chunked = RenameFrame(chunked, map_columns)
            split_by = [map_columns.get(c, c) for c in split_by]

        # Sort or shuffle
        split_every = getattr(self, "split_every", 0) or chunked.npartitions
        ignore_index = getattr(self, "ignore_index", True)
        if self.shuffle_by_index is not None:
            ignore_index = not self.shuffle_by_index