Class _HLGExprSequence

dask/_expr.py:1086–1187

class _HLGExprSequence(Expr):

    def __getitem__(self, other):
        return self.operands[other]

    def _operands_for_repr(self):
        return [
            f"name={self.operand('name')!r}",
            f"dsk={self.operand('dsk')!r}",
        ]

    def _tree_repr_lines(self, indent=0, recursive=True):
        return self._operands_for_repr()

    def finalize_compute(self):
        return _HLGExprSequence(*[op.finalize_compute() for op in self.operands])

    def _tune_down(self):
        if len(self.operands) == 1:
            return None
        from dask.highlevelgraph import HighLevelGraph

        groups = toolz.groupby(
            lambda x: x.low_level_optimizer if isinstance(x, HLGExpr) else None,
            self.operands,
        )
        exprs = []
        changed = False
        for optimizer, group in groups.items():
            if len(group) > 1:
                graphs = [expr.hlg for expr in group]

                changed = True
                dsk = HighLevelGraph.merge(*graphs)
                hlg_group = _HLGExprGroup(
                    dsk=dsk,
                    low_level_optimizer=optimizer,
                    output_keys=[v.__dask_keys__() for v in group],
                    postcompute=[g.postcompute for g in group],
                )
                exprs.append(hlg_group)
            else:
                exprs.append(group[0])
        if not changed:
            return None
        return _HLGExprSequence(*exprs)

    @functools.cached_property
    def _optimized_dsk(self) -> HighLevelGraph:
        from dask.highlevelgraph import HighLevelGraph

        hlgexpr: HLGExpr
        graphs = []
        # simplify_down ensure there are only one HLGExpr per optimizer/finalizer
        for hlgexpr in self.operands:
            keys = hlgexpr.__dask_keys__()
            dsk = hlgexpr.hlg
            if (optimizer := hlgexpr.low_level_optimizer) is not None: