Method cull

dask/highlevelgraph.py:127–182

Remove unnecessary tasks from the layer In other words, return a new Layer with only the tasks required to calculate `keys` and a map of external key dependencies. Examples -------- >>> inc = lambda x: x + 1 >>> add = lambda x, y: x + y >>> d

(
        self, keys: set[Key], all_hlg_keys: Collection[Key]
    )

        return self.keys()  # this implementation will materialize the graph

    def cull(
        self, keys: set[Key], all_hlg_keys: Collection[Key]
    ) -> tuple[Layer, Mapping[Key, set[Key]]]:
        """Remove unnecessary tasks from the layer

        In other words, return a new Layer with only the tasks required to
        calculate `keys` and a map of external key dependencies.

        Examples
        --------
        >>> inc = lambda x: x + 1
        >>> add = lambda x, y: x + y
        >>> d = MaterializedLayer({'x': 1, 'y': (inc, 'x'), 'out': (add, 'x', 10)})
        >>> _, deps = d.cull({'out'}, d.keys())
        >>> deps
        {'out': {'x'}, 'x': set()}

        Returns
        -------
        layer: Layer
            Culled layer
        deps: Map
            Map of external key dependencies
        """

        if self.has_legacy_tasks:
            if len(keys) == len(self):
                # Nothing to cull if preserving all existing keys
                return (
                    self,
                    {k: self.get_dependencies(k, all_hlg_keys) for k in self.keys()},
                )
            ret_deps = {}
            seen = set()
            out = {}
            work = keys.copy()
            while work:
                k = work.pop()
                if k not in self:
                    continue
                out[k] = self[k]
                ret_deps[k] = self.get_dependencies(k, all_hlg_keys)
                for d in ret_deps[k]:
                    if d not in seen:
                        if d in self:
                            seen.add(d)
                            work.add(d)

            return MaterializedLayer(out, annotations=self.annotations), ret_deps
        else:
            from dask._task_spec import cull

            out = cull(dict(self), keys)
            return MaterializedLayer(out, annotations=self.annotations), {
                k: set(v.dependencies) for k, v in out.items()
            }

    def get_dependencies(self, key: Key, all_hlg_keys: Collection[Key]) -> set: