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

Code before:
    def analyze(self, filename: str | None = None, format: str | None = None) -> None:
        from dask.dataframe.dask_expr._expr import Expr as DFExpr
        from dask.dataframe.dask_expr.diagnostics import analyze

        if not isinstance(self, DFExpr):
            raise TypeError(
                "analyze is only supported for dask.dataframe.Expr objects."
            )
        return analyze(self, filename=filename, format=format)

    def explain(
        self, stage: OptimizerStage = "fused", format: str | None = None
    ) -> None:
        from dask.dataframe.dask_expr.diagnostics import explain

        return explain(self, stage, format)

    def pprint(self):
        for line in self._tree_repr_lines():
            print(line)

    def __hash__(self):
        return hash(self._name)

    def __dask_tokenize__(self):
        if not self._determ_token:
            # If the subclass does not implement a __dask_tokenize__ we'll want
            # to tokenize all operands.
            # Note how this differs to the implementation of
            # Expr.deterministic_token
            self._determ_token = _tokenize_deterministic(self.operands)
        return self._determ_token

    @staticmethod
    def _reconstruct(*args):
        typ, *operands, token = args
        return typ(*operands, _determ_token=token)

    def __reduce__(self):
        if dask.config.get("dask-expr-no-serialize", False):
            raise RuntimeError(f"Serializing a {type(self)} object")
        return Expr._reconstruct, tuple(
            [type(self), *self.operands, self.deterministic_token]
        )

    def _depth(self, cache=None):
        """Depth of the expression tree

        Returns
        -------
        depth: int

Review comment (suggestion):
On main

`_name` -> `deterministic_token` -> `token(self.operands)` -> HLG tokenization is not reliable (see https://github.com/dask/dask/blob/e0877d0a89622adf3fb68db3d7ec43475a253fd9/dask/highlevelgraph.py#L999-L1003 / https://github.com/dask/dask/issues/9888)

This PR

`_name` -> `deterministic_token` -> `__dask_tokenize__` -> `id(self)`

Code after:
    def analyze(self, filename: str | None = None, format: str | None = None) -> None:
        from dask.dataframe.dask_expr._expr import Expr as DFExpr
        from dask.dataframe.dask_expr.diagnostics import analyze

        if not isinstance(self, DFExpr):
            raise TypeError(
                "analyze is only supported for dask.dataframe.Expr objects."
            )
        return analyze(self, filename=filename, format=format)

    def explain(
        self, stage: OptimizerStage = "fused", format: str | None = None
    ) -> None:
        from dask.dataframe.dask_expr.diagnostics import explain

        return explain(self, stage, format)

    def pprint(self):
        for line in self._tree_repr_lines():
            print(line)

    def __hash__(self):
        return hash(self._name)

    def __dask_tokenize__(self):
        if not self._determ_token:
            # If the subclass does not implement a __dask_tokenize__ we'll want
            # to tokenize all operands.
            # Note how this differs to the implementation of
            # Expr.deterministic_token
            self._determ_token = _tokenize_deterministic(type(self), *self.operands)
        return self._determ_token

    @staticmethod
    def _reconstruct(*args):
        typ, *operands, token = args
        return typ(*operands, _determ_token=token)

    def __reduce__(self):
        if dask.config.get("dask-expr-no-serialize", False):
            raise RuntimeError(f"Serializing a {type(self)} object")
        return Expr._reconstruct, tuple(
            [type(self), *self.operands, self.deterministic_token]
        )

    def _depth(self, cache=None):
        """Depth of the expression tree

        Returns
        -------
        depth: int