fix: include class-based UDF instance state in hash

docs/guide/checkpoints.md

@@ -202,3 +202,4 @@ dc.read_storage("gs://datachain-demo/dogs-and-cats/", anon=True).map(
 - **Script path matters:** DataChain links runs by the script's absolute path. Moving the script breaks checkpoint linking.
 - **Threading/multiprocessing:** Checkpoints are automatically disabled when Python threading or multiprocessing is detected. DataChain's built-in `parallel` setting for Python operations is not affected.
 - **Unhashable callables:** Built-in functions (`len`, `str`), C extensions, and `Mock` objects produce a different hash on each run, so checkpoints using these as Python operations will always recompute. Use regular `def` functions or lambdas instead.
+- **Class UDF state:** Constructor callables and custom objects recompute by default. Override [`state_hash()`](python-engine.md#caching-class-based-operations) only when you can provide a complete stable identity.

docs/guide/python-engine.md

@@ -151,6 +151,40 @@ class ImageEncoder(Mapper):
         del self.model
 ```
 
+### Caching class-based operations
+
+DataChain hashes UDF code, schemas, and constructor arguments. Primitive values and
+nested built-in containers are handled automatically. Callables and custom objects
+receive a unique identity instead, preventing incorrect cache reuse. Override
+`state_hash()` when you can identify such state safely:
+
+```python
+import hashlib
+import json
+from datachain.lib.udf import Mapper
+
+class Tokenize(Mapper):
+    def __init__(self, tokenizer, tokenizer_id: str, max_length: int):
+        self.tokenizer = tokenizer
+        self.tokenizer_id = tokenizer_id
+        self.max_length = max_length
+
+    def state_hash(self) -> str:
+        state = json.dumps(
+            {"tokenizer_id": self.tokenizer_id, "max_length": self.max_length},
+            sort_keys=True,
+        )
+        return hashlib.sha256(state.encode()).hexdigest()
+
+    def process(self, text: str) -> list[str]:
+        return self.tokenizer(text)[: self.max_length]
+```
+
+`state_hash()` must return a SHA-256 hexadecimal string covering all instance state
+that affects output. When all constructor arguments are supported primitives, call
+`super().state_hash()` to include them in an override. UDF code and schemas are always
+included; an incomplete hash can reuse an incorrect cached result.
+
 Use class-based operations sparingly; `.setup()` covers most cases.
 
 ## Execution and Scale

docs/references/llm.md

@@ -139,7 +139,8 @@ Reliability is layered:
 Materialized `llm.*` columns are cached and versioned, so re-running a chain reads
 the stored result instead of re-calling the model; the cache invalidates when any
 output-affecting input changes (model, prompt, schema, the input column, `type`,
-params, ...).
+params, ...). Custom objects in parameters receive a per-operation identity, safely
+disabling reuse across separately created operations.
 
 ## No fused predicate
 

src/datachain/hash_utils.py

@@ -1,6 +1,7 @@
 import hashlib
 import inspect
 import logging
+import struct
 import textwrap
 from collections.abc import Sequence
 from typing import TypeAlias, TypeVar
@@ -16,6 +17,46 @@
 ColumnLike: TypeAlias = str | T
 
 
+def normalize_hash_value(value, *, sort_dicts: bool = False):  # noqa: PLR0911
+    """Return a complete, deterministic representation for stable cache keys."""
+    value_type = type(value)
+
+    def normalize(item):
+        return normalize_hash_value(item, sort_dicts=sort_dicts)
+
+    if value is None:
+        return ("none",)
+    if value_type is float:
+        return ("float", struct.pack("!d", value))
+    if value_type in (bool, int, str, bytes):
+        return (value_type.__name__, value)
+    if value_type is dict:
+        items = tuple((normalize(key), normalize(item)) for key, item in value.items())
+        return ("dict", tuple(sorted(items, key=repr)) if sort_dicts else items)
+    if value_type is set:
+        return (
+            "set",
+            tuple(sorted(map(normalize, value), key=repr)),
+        )
+    if value_type is frozenset:
+        return (
+            "frozenset",
+            tuple(sorted(map(normalize, value), key=repr)),
+        )
+    if value_type is list:
+        return ("list", tuple(map(normalize, value)))
+    if value_type is tuple:
+        return ("tuple", tuple(map(normalize, value)))
+
+    raise TypeError(f"value of type {value_type.__name__!r} cannot be hashed safely")
+
+
+def hash_value(value) -> str:
+    """Hash a value after normalizing unordered containers."""
+    normalized = normalize_hash_value(value)
+    return hashlib.sha256(repr(normalized).encode("utf-8")).hexdigest()
+
+
 def _serialize_value(val):  # noqa: PLR0911
     """Helper to serialize arbitrary values recursively."""
     if val is None:

src/datachain/lib/arrow.py

@@ -1,4 +1,5 @@
 import math
+import pickle
 from collections.abc import Sequence
 from itertools import islice
 from typing import TYPE_CHECKING, Any
@@ -33,6 +34,27 @@
 DATACHAIN_SIGNAL_SCHEMA_PARQUET_KEY = b"DataChain SignalSchema"
 
 
+def _parse_options_hash_args(options: ParseOptions) -> dict[str, Any]:
+    return {
+        "delimiter": options.delimiter,
+        "double_quote": options.double_quote,
+        "escape_char": options.escape_char,
+        "ignore_empty_lines": options.ignore_empty_lines,
+        "invalid_row_handler": options.invalid_row_handler,
+        "newlines_in_values": options.newlines_in_values,
+        "quote_char": options.quote_char,
+    }
+
+
+def _csv_format_hash_args(format: CsvFileFormat) -> dict[str, Any]:
+    scan_options = format.default_fragment_scan_options
+    return {
+        "parse_options": _parse_options_hash_args(format.parse_options),
+        "read_options": pickle.dumps(scan_options.read_options),
+        "convert_options": pickle.dumps(scan_options.convert_options),
+    }
+
+
 def fix_pyarrow_format(format, parse_options=None):
     # Re-init invalid row handler: https://issues.apache.org/jira/browse/ARROW-17641
     if (
@@ -54,6 +76,7 @@ def __init__(
         output_schema: type["BaseModel"] | None = None,
         source: bool = True,
         nrows: int | None = None,
+        _generated_output_schema: bool = False,
         **kwargs,
     ):
         """
@@ -75,6 +98,25 @@ def __init__(
         self.parse_options = kwargs.pop("parse_options", None)
         self.kwargs = kwargs
 
+    @classmethod
+    def _constructor_hash_args(cls, arguments):
+        arguments = arguments.copy()
+        generated_output_schema = arguments.pop("_generated_output_schema")
+        if generated_output_schema:
+            # Its stable field shape is already part of the output signal schema.
+            arguments.pop("output_schema", None)
+        input_schema = arguments.get("input_schema")
+        if input_schema is not None:
+            arguments["input_schema"] = input_schema.serialize().to_pybytes()
+        kwargs = arguments["kwargs"].copy()
+        for name, value in kwargs.items():
+            if isinstance(value, ParseOptions):
+                kwargs[name] = _parse_options_hash_args(value)
+            elif isinstance(value, CsvFileFormat):
+                kwargs[name] = _csv_format_hash_args(value)
+        arguments["kwargs"] = kwargs
+        return arguments
+
     def process(self, file: File):
         if file._caching_enabled:
             file.ensure_cached()

src/datachain/lib/dc/datachain.py

@@ -2537,7 +2537,9 @@ def parse_tabular(
             except ValueError as e:
                 raise DatasetPrepareError(self.name, e) from e
 
+        generated_output_schema = False
         if isinstance(output, dict):
+            generated_output_schema = True
             model_name = model_name or column or ""
             model = dict_to_data_model(model_name, output)
             output = model
@@ -2563,6 +2565,7 @@ def parse_tabular(
                 model,
                 source,
                 nrows,
+                _generated_output_schema=generated_output_schema,
                 parse_options=parse_options,
                 **kwargs,
             ),

src/datachain/lib/hf.py

@@ -46,6 +46,19 @@
 )
 
 
+def _dataset_hash_args(ds: HFDatasetType) -> Any:
+    if isinstance(ds, (DatasetDict, IterableDatasetDict)):
+        return (
+            type(ds).__name__,
+            {name: _dataset_hash_args(split) for name, split in ds.items()},
+        )
+    if isinstance(ds, (Dataset, IterableDataset)):
+        fingerprint = getattr(ds, "_fingerprint", None)
+        if fingerprint is not None:
+            return (type(ds).__name__, fingerprint)
+    return ds
+
+
 class HFClassLabel(DataModel):
     string: str
     integer: int
@@ -93,6 +106,17 @@ def __init__(
         self.args = args
         self.kwargs = kwargs
 
+    @classmethod
+    def _constructor_hash_args(cls, arguments):
+        # output_schema is a dynamically-created pydantic class with a random
+        # name suffix; its stable field shape lands in the UDF hash via the
+        # output signal schema, so drop it here to keep the constructor hash
+        # deterministic across runs.
+        arguments = arguments.copy()
+        arguments.pop("output_schema", None)
+        arguments["ds"] = _dataset_hash_args(arguments["ds"])
+        return arguments
+
     def setup(self):
         self.ds_dict = stream_splits(self.ds, *self.args, **self.kwargs)
 

src/datachain/lib/udf.py

@@ -8,6 +8,7 @@
 from functools import partial
 from graphlib import CycleError, TopologicalSorter
 from typing import TYPE_CHECKING, Any, TypeVar
+from uuid import uuid4
 
 import attrs
 from fsspec.callbacks import DEFAULT_CALLBACK, Callback
@@ -16,7 +17,7 @@
 from datachain.asyn import AsyncMapper
 from datachain.cache import temporary_cache
 from datachain.dataset import RowDict
-from datachain.hash_utils import hash_callable
+from datachain.hash_utils import hash_callable, hash_value
 from datachain.lib.convert.flatten import (
     classify_field,
     flatten,
@@ -223,6 +224,20 @@ def prefetch(self) -> int:
         return self.inner.prefetch
 
 
+def _hash_constructor_args(
+    arguments: dict[str, Any], *, warn_on_unsupported: bool = True
+) -> str:
+    try:
+        return hash_value(arguments)
+    except (TypeError, RecursionError) as exc:
+        if warn_on_unsupported:
+            logger.warning(
+                "%s; cache reuse across UDF instances is disabled",
+                exc,
+            )
+        return hashlib.sha256(uuid4().bytes).hexdigest()
+
+
 class UDFBase(AbstractUDF):
     """Base class for stateful user-defined functions.
 
@@ -274,11 +289,49 @@ def process(self, file) -> list[float]:
     is_input_batched = False
     is_output_batched = False
     prefetch: int = 0
+    # Set in __new__ from bound constructor args (see hash()); default keeps
+    # mypy happy and acts as a safe fallback for instances that skip __new__.
+    _constructor_state_hash: str = ""
     # Class-level default so subclasses that skip super().__init__() still
     # have this attribute; _MultiSignalMapper sets params/output per entry
     # but doesn't call _init(), so _func would otherwise be missing.
     _func: Callable | None = None
 
+    def __new__(cls, *args, **kwargs):
+        instance = super().__new__(cls)
+        try:
+            # bound-method signature so `self` is already stripped; correct even
+            # when __init__ uses *args instead of a named `self` parameter.
+            bound = inspect.signature(instance.__init__).bind(*args, **kwargs)
+        except TypeError:
+            # Pickle allocates an empty instance and restores its attributes
+            # afterwards, without calling __init__ or supplying its arguments.
+            return instance
+
+        bound.apply_defaults()
+        arguments = cls._constructor_hash_args(dict(bound.arguments))
+        instance._constructor_state_hash = _hash_constructor_args(
+            arguments,
+            warn_on_unsupported=cls.state_hash is UDFBase.state_hash,
+        )
+        return instance
+
+    @classmethod
+    def _constructor_hash_args(cls, arguments: dict[str, Any]) -> dict[str, Any]:
+        """Constructor arguments that determine this UDF instance's identity."""
+        return arguments
+
+    def state_hash(self) -> str:
+        """Return a stable SHA-256 hash for state that affects this UDF's output.
+
+        Override this when constructor arguments contain callables or other opaque
+        objects that DataChain cannot hash safely. The method is called after
+        ``__init__`` and must account for all per-instance behavioral state. By
+        default, DataChain returns its automatic constructor-argument hash. Call
+        ``super().state_hash()`` to include that hash in an override.
+        """
+        return self._constructor_state_hash
+
     def __init__(self):
         self.params: SignalSchema | None = None
         self.output = None
@@ -304,6 +357,25 @@ def hash(self, include_body: bool = True) -> str:
             self.params.hash() if self.params else "",
             self.output.hash(),
         ]
+        # For class-based UDFs, mix in constructor state so two instances that
+        # differ only in constructor args don't collide.
+        if self._func is None:
+            state_hash = self.state_hash()
+            if not isinstance(state_hash, str) or len(state_hash) != 64:
+                raise ValueError(
+                    "state_hash() must return a SHA-256 hexadecimal string"
+                )
+            try:
+                state_hash_bytes = bytes.fromhex(state_hash)
+            except ValueError as exc:
+                raise ValueError(
+                    "state_hash() must return a SHA-256 hexadecimal string"
+                ) from exc
+            if len(state_hash_bytes) != 32:
+                raise ValueError(
+                    "state_hash() must return a SHA-256 hexadecimal string"
+                )
+            parts.append(state_hash)
 
         return hashlib.sha256(
             b"".join([bytes.fromhex(part) for part in parts])
@@ -596,6 +668,12 @@ class _MultiSignalMapper(Mapper):
     irrelevant. Cycles raise ``ValueError`` at construction time.
     """
 
+    @classmethod
+    def _constructor_hash_args(cls, arguments):
+        # hash() is fully overridden below; skip the base state hash so we
+        # don't warn on the callables in signal_map or store unused bytes.
+        return {}
+
     def __init__(
         self,
         signal_map: "dict[str, Callable | UDFBase]",