From a4b872252dcabbbd62b484112d3b67518fcd44f9 Mon Sep 17 00:00:00 2001
From: ilongin <ivan.longin1@gmail.com>
Date: Tue, 18 Aug 2026 02:25:00 +0200
Subject: [PATCH 01/16] refactoring

---
 src/datachain/lib/arrow.py          | 14 +++++++++++
 src/datachain/lib/hf.py             | 13 ++++++++++
 src/datachain/lib/udf.py            | 15 ++++++++++-
 tests/func/test_udf.py              | 39 +++++++++++++++++++++++++++++
 tests/unit/lib/test_udf.py          | 16 ++++++++++++
 tests/unit/test_query_steps_hash.py |  4 +--
 6 files changed, 98 insertions(+), 3 deletions(-)

diff --git a/src/datachain/lib/arrow.py b/src/datachain/lib/arrow.py
index 6cbb18f1a..1041004c3 100644
--- a/src/datachain/lib/arrow.py
+++ b/src/datachain/lib/arrow.py
@@ -21,6 +21,7 @@
 from datachain.lib.udf import Generator
 from datachain.lib.utils import normalize_col_names
 from datachain.progress import tqdm
+from datachain.utils import filtered_cloudpickle_dumps
 
 if TYPE_CHECKING:
     from datasets.features.features import Features
@@ -75,6 +76,19 @@ def __init__(
         self.parse_options = kwargs.pop("parse_options", None)
         self.kwargs = kwargs
 
+    def _hash_state(self) -> bytes:
+        # output_schema is a dynamically-created pydantic class with a random
+        # name suffix; its stable field shape is already in self.output.hash().
+        return filtered_cloudpickle_dumps(
+            {
+                "input_schema": self.input_schema,
+                "source": self.source,
+                "nrows": self.nrows,
+                "parse_options": self.parse_options,
+                "kwargs": self.kwargs,
+            }
+        )
+
     def process(self, file: File):
         if file._caching_enabled:
             file.ensure_cached()
diff --git a/src/datachain/lib/hf.py b/src/datachain/lib/hf.py
index 3f6db377a..a6601aeb5 100644
--- a/src/datachain/lib/hf.py
+++ b/src/datachain/lib/hf.py
@@ -35,6 +35,7 @@
 from datachain.lib.udf import Generator
 from datachain.lib.utils import normalize_col_names
 from datachain.progress import tqdm
+from datachain.utils import filtered_cloudpickle_dumps
 
 if TYPE_CHECKING:
     import pyarrow as pa
@@ -93,6 +94,18 @@ def __init__(
         self.args = args
         self.kwargs = kwargs
 
+    def _hash_state(self) -> bytes:
+        # output_schema is a dynamically-created pydantic class with a random
+        # name suffix; its stable field shape is already in self.output.hash().
+        return filtered_cloudpickle_dumps(
+            {
+                "ds": self.ds,
+                "limit": self.limit,
+                "args": self.args,
+                "kwargs": self.kwargs,
+            }
+        )
+
     def setup(self):
         self.ds_dict = stream_splits(self.ds, *self.args, **self.kwargs)
 
diff --git a/src/datachain/lib/udf.py b/src/datachain/lib/udf.py
index e112b975a..97a9b64d2 100644
--- a/src/datachain/lib/udf.py
+++ b/src/datachain/lib/udf.py
@@ -32,7 +32,7 @@
     Partition,
     RowsOutputBatch,
 )
-from datachain.utils import safe_closing, with_last_flag
+from datachain.utils import filtered_cloudpickle_dumps, safe_closing, with_last_flag
 
 logger = logging.getLogger("datachain")
 
@@ -252,11 +252,24 @@ def hash(self, include_body: bool = True) -> str:
             self.params.hash() if self.params else "",
             self.output.hash(),
         ]
+        # For class-based UDFs, mix in instance state so two instances that
+        # differ only in constructor args don't collide.
+        if self._func is None and (state := self._hash_state()) is not None:
+            parts.append(hashlib.sha256(state).hexdigest())
 
         return hashlib.sha256(
             b"".join([bytes.fromhex(part) for part in parts])
         ).hexdigest()
 
+    def _hash_state(self) -> bytes | None:
+        """State bytes to mix into hash() for class-based UDFs. Default is a
+        cloudpickle of the whole instance. Subclasses can override when the
+        default contains non-deterministic data (e.g. dynamically-named
+        pydantic classes) or when they want to declare identity explicitly.
+        Return None to skip the state part entirely.
+        """
+        return filtered_cloudpickle_dumps(self)
+
     def process(self, *args, **kwargs):
         """Processing function that needs to be defined by user"""
         if not self._func:
diff --git a/tests/func/test_udf.py b/tests/func/test_udf.py
index 6454e15ea..bd5887c74 100644
--- a/tests/func/test_udf.py
+++ b/tests/func/test_udf.py
@@ -8,6 +8,7 @@
 
 import multiprocess as mp
 import pytest
+from pydantic import BaseModel
 
 import datachain as dc
 from datachain.client.fileslice import FileWrapper
@@ -256,6 +257,44 @@ def process(self, size):
     ]
 
 
+def test_class_agg_instance_state_produces_distinct_results(test_session):
+    class Out(BaseModel):
+        key: int
+        n: int
+
+    class CountAbove(dc.Aggregator):
+        def __init__(self, limit: int):
+            super().__init__()
+            self.limit = limit
+
+        def process(self, key, value):
+            yield Out(key=key[0], n=sum(1 for v in value if v > self.limit))
+
+    src = dc.read_values(
+        key=[1] * 5 + [2] * 5,
+        value=[1, 2, 3, 4, 5, 1, 2, 3, 4, 5],
+        session=test_session,
+    )
+    rows_zero = sorted(
+        src.agg(
+            CountAbove(0),
+            partition_by="key",
+            params=["key", "value"],
+            output={"o": Out},
+        ).to_list("o.key", "o.n")
+    )
+    rows_three = sorted(
+        src.agg(
+            CountAbove(3),
+            partition_by="key",
+            params=["key", "value"],
+            output={"o": Out},
+        ).to_list("o.key", "o.n")
+    )
+    assert rows_zero == [(1, 5), (2, 5)]
+    assert rows_three == [(1, 2), (2, 2)]
+
+
 @pytest.mark.parametrize(
     "cloud_type,version_aware",
     [("s3", True)],
diff --git a/tests/unit/lib/test_udf.py b/tests/unit/lib/test_udf.py
index d2100a4a0..ba7790c61 100644
--- a/tests/unit/lib/test_udf.py
+++ b/tests/unit/lib/test_udf.py
@@ -135,6 +135,22 @@ def test_udf_verbose_name_unknown():
     assert udf.verbose_name == "<unknown>"
 
 
+def test_class_udf_hash_varies_with_instance_state():
+    class Limited(Mapper):
+        def __init__(self, limit: int):
+            super().__init__()
+            self.limit = limit
+
+        def process(self, x: int) -> int:
+            return x + self.limit
+
+    sign_a = get_sign(Limited(0), output="y")
+    sign_b = get_sign(Limited(3), output="y")
+    udf_a = Mapper._create(sign_a, sign_a.output_schema)
+    udf_b = Mapper._create(sign_b, sign_b.output_schema)
+    assert udf_a.hash() != udf_b.hash()
+
+
 def test_udf_does_not_traverse_setup_value():
     value = {}
     value["self"] = value
diff --git a/tests/unit/test_query_steps_hash.py b/tests/unit/test_query_steps_hash.py
index b5f24d1a5..dad22e220 100644
--- a/tests/unit/test_query_steps_hash.py
+++ b/tests/unit/test_query_steps_hash.py
@@ -322,7 +322,7 @@ def test_subtract_hash(test_session, numbers_dataset, on):
             DoubleMapper(),
             ["x"],
             {"double": int},
-            "b58c9679ed454d3f54b4a754585727697a9aea9e4725bd12a842a774b5087963",
+            "a91d1f4093958aff06588e39b69892eb108d22c452717d15740b271bfd3a6157",
         ),
     ],
 )
@@ -362,7 +362,7 @@ def test_udf_mapper_hash(
             TripleGenerator(),
             ["x"],
             {"triple": int},
-            "01201327b1926788e6242d2be5383c63b97ec018232ab0844f047cf64ec2dfca",
+            "b158fe7fb7396352ba1c4e47a92f70b3ec261228eb0968292340fb75ab60eea1",
         ),
     ],
 )

From 47dff658cfde03d4c3b876911a0bf152c326b7a2 Mon Sep 17 00:00:00 2001
From: ilongin <ivan.longin1@gmail.com>
Date: Tue, 18 Aug 2026 02:30:11 +0200
Subject: [PATCH 02/16] added test

---
 tests/unit/lib/test_udf.py | 16 ++++++++++++++++
 1 file changed, 16 insertions(+)

diff --git a/tests/unit/lib/test_udf.py b/tests/unit/lib/test_udf.py
index ba7790c61..4e090d1cd 100644
--- a/tests/unit/lib/test_udf.py
+++ b/tests/unit/lib/test_udf.py
@@ -151,6 +151,22 @@ def process(self, x: int) -> int:
     assert udf_a.hash() != udf_b.hash()
 
 
+def test_class_udf_hash_is_deterministic_across_instances():
+    class Limited(Mapper):
+        def __init__(self, limit: int):
+            super().__init__()
+            self.limit = limit
+
+        def process(self, x: int) -> int:
+            return x + self.limit
+
+    sign_a = get_sign(Limited(3), output="y")
+    sign_b = get_sign(Limited(3), output="y")
+    udf_a = Mapper._create(sign_a, sign_a.output_schema)
+    udf_b = Mapper._create(sign_b, sign_b.output_schema)
+    assert udf_a.hash() == udf_b.hash()
+
+
 def test_udf_does_not_traverse_setup_value():
     value = {}
     value["self"] = value

From 34b5a90277eefbfc2a8b53f41627eae96162c468 Mon Sep 17 00:00:00 2001
From: ilongin <ivan.longin1@gmail.com>
Date: Wed, 19 Aug 2026 14:24:10 +0200
Subject: [PATCH 03/16] refactoring hashing

---
 src/datachain/hash_utils.py         |  44 ++++++++++++
 src/datachain/lib/arrow.py          |  20 +++---
 src/datachain/lib/hf.py             |  19 +++--
 src/datachain/lib/udf.py            |  54 ++++++++++----
 src/datachain/llm/spec.py           |  31 +-------
 tests/unit/lib/test_arrow.py        |  12 ++++
 tests/unit/lib/test_hf.py           |  13 ++++
 tests/unit/lib/test_udf.py          | 105 ++++++++++++++++++++++++++++
 tests/unit/test_hash_utils.py       |  41 ++++++++++-
 tests/unit/test_query_steps_hash.py |   4 +-
 10 files changed, 275 insertions(+), 68 deletions(-)

diff --git a/src/datachain/hash_utils.py b/src/datachain/hash_utils.py
index af90fcd94..d1c71e503 100644
--- a/src/datachain/hash_utils.py
+++ b/src/datachain/hash_utils.py
@@ -1,7 +1,9 @@
 import hashlib
 import inspect
 import logging
+import re
 import textwrap
+import warnings
 from collections.abc import Sequence
 from typing import TypeAlias, TypeVar
 from uuid import uuid4
@@ -16,6 +18,48 @@
 ColumnLike: TypeAlias = str | T
 
 
+# A value with no custom repr shows its memory address (``<Cls object at 0x...>``
+# for plain objects, ``<function name at 0x...>`` for functions and methods),
+# which changes each run and would make the cache key unstable.
+_DEFAULT_REPR = re.compile(r" at 0x[0-9a-fA-F]+>")
+
+
+def normalize_hash_value(value):
+    """Order-independent form of a value for stable cache keys."""
+    if isinstance(value, dict):
+        items = (
+            (normalize_hash_value(k), normalize_hash_value(v)) for k, v in value.items()
+        )
+        return ("dict", tuple(sorted(items, key=repr)))
+    if isinstance(value, set):
+        return (
+            "set",
+            tuple(sorted(map(normalize_hash_value, value), key=repr)),
+        )
+    if isinstance(value, frozenset):
+        return (
+            "frozenset",
+            tuple(sorted(map(normalize_hash_value, value), key=repr)),
+        )
+    if isinstance(value, list):
+        return ("list", tuple(map(normalize_hash_value, value)))
+    if isinstance(value, tuple):
+        return ("tuple", tuple(map(normalize_hash_value, value)))
+    if _DEFAULT_REPR.search(repr(value)):
+        warnings.warn(
+            f"value of type {type(value).__name__!r} has no stable repr; it "
+            "breaks caching (full recompute every run).",
+            stacklevel=2,
+        )
+    return value
+
+
+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:
diff --git a/src/datachain/lib/arrow.py b/src/datachain/lib/arrow.py
index 1041004c3..e15e0cbdf 100644
--- a/src/datachain/lib/arrow.py
+++ b/src/datachain/lib/arrow.py
@@ -21,7 +21,6 @@
 from datachain.lib.udf import Generator
 from datachain.lib.utils import normalize_col_names
 from datachain.progress import tqdm
-from datachain.utils import filtered_cloudpickle_dumps
 
 if TYPE_CHECKING:
     from datasets.features.features import Features
@@ -76,18 +75,15 @@ def __init__(
         self.parse_options = kwargs.pop("parse_options", None)
         self.kwargs = kwargs
 
-    def _hash_state(self) -> bytes:
+    @classmethod
+    def _constructor_hash_args(cls, arguments):
         # output_schema is a dynamically-created pydantic class with a random
-        # name suffix; its stable field shape is already in self.output.hash().
-        return filtered_cloudpickle_dumps(
-            {
-                "input_schema": self.input_schema,
-                "source": self.source,
-                "nrows": self.nrows,
-                "parse_options": self.parse_options,
-                "kwargs": self.kwargs,
-            }
-        )
+        # 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)
+        return arguments
 
     def process(self, file: File):
         if file._caching_enabled:
diff --git a/src/datachain/lib/hf.py b/src/datachain/lib/hf.py
index a6601aeb5..0cdf55395 100644
--- a/src/datachain/lib/hf.py
+++ b/src/datachain/lib/hf.py
@@ -35,7 +35,6 @@
 from datachain.lib.udf import Generator
 from datachain.lib.utils import normalize_col_names
 from datachain.progress import tqdm
-from datachain.utils import filtered_cloudpickle_dumps
 
 if TYPE_CHECKING:
     import pyarrow as pa
@@ -94,17 +93,15 @@ def __init__(
         self.args = args
         self.kwargs = kwargs
 
-    def _hash_state(self) -> bytes:
+    @classmethod
+    def _constructor_hash_args(cls, arguments):
         # output_schema is a dynamically-created pydantic class with a random
-        # name suffix; its stable field shape is already in self.output.hash().
-        return filtered_cloudpickle_dumps(
-            {
-                "ds": self.ds,
-                "limit": self.limit,
-                "args": self.args,
-                "kwargs": self.kwargs,
-            }
-        )
+        # 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)
+        return arguments
 
     def setup(self):
         self.ds_dict = stream_splits(self.ds, *self.args, **self.kwargs)
diff --git a/src/datachain/lib/udf.py b/src/datachain/lib/udf.py
index 97a9b64d2..e3c056137 100644
--- a/src/datachain/lib/udf.py
+++ b/src/datachain/lib/udf.py
@@ -16,7 +16,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,
@@ -32,7 +32,7 @@
     Partition,
     RowsOutputBatch,
 )
-from datachain.utils import filtered_cloudpickle_dumps, safe_closing, with_last_flag
+from datachain.utils import safe_closing, with_last_flag
 
 logger = logging.getLogger("datachain")
 
@@ -226,6 +226,35 @@ 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 = ""
+
+    def __new__(cls, *args, **kwargs):
+        instance = super().__new__(cls)
+        signature = inspect.signature(cls.__init__)
+
+        try:
+            bound = signature.bind(instance, *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 = dict(bound.arguments)
+        if signature.parameters:
+            self_name = next(iter(signature.parameters))
+            arguments.pop(self_name, None)
+
+        arguments = cls._constructor_hash_args(arguments)
+        instance._constructor_state_hash = hash_value(arguments)
+        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 __init__(self):
         self.params: SignalSchema | None = None
@@ -252,24 +281,15 @@ def hash(self, include_body: bool = True) -> str:
             self.params.hash() if self.params else "",
             self.output.hash(),
         ]
-        # For class-based UDFs, mix in instance state so two instances that
+        # 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 and (state := self._hash_state()) is not None:
-            parts.append(hashlib.sha256(state).hexdigest())
+        if self._func is None:
+            parts.append(self._constructor_state_hash)
 
         return hashlib.sha256(
             b"".join([bytes.fromhex(part) for part in parts])
         ).hexdigest()
 
-    def _hash_state(self) -> bytes | None:
-        """State bytes to mix into hash() for class-based UDFs. Default is a
-        cloudpickle of the whole instance. Subclasses can override when the
-        default contains non-deterministic data (e.g. dynamically-named
-        pydantic classes) or when they want to declare identity explicitly.
-        Return None to skip the state part entirely.
-        """
-        return filtered_cloudpickle_dumps(self)
-
     def process(self, *args, **kwargs):
         """Processing function that needs to be defined by user"""
         if not self._func:
@@ -547,6 +567,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]):
         super().__init__()
         self._signal_map = signal_map
diff --git a/src/datachain/llm/spec.py b/src/datachain/llm/spec.py
index 1d4022a8a..99d631206 100644
--- a/src/datachain/llm/spec.py
+++ b/src/datachain/llm/spec.py
@@ -1,11 +1,10 @@
-import re
-import warnings
 from collections.abc import Callable, Iterator
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any, Literal, get_args, get_origin
 
 from pydantic import BaseModel
 
+from datachain.hash_utils import normalize_hash_value
 from datachain.lib.udf import BindContext, BoundSpec
 from datachain.llm import engine
 from datachain.llm.content import MEDIA_VALUES, Media, build_messages, to_text
@@ -27,30 +26,6 @@ def _element_type(schema: Any) -> tuple[Any, bool]:
     return schema, False
 
 
-# A value with no custom repr shows its memory address (``<Cls object at 0x...>``),
-# which changes each run and would make the cache key unstable.
-_DEFAULT_REPR = re.compile(r" object at 0x[0-9a-fA-F]+>")
-
-
-def _canonical(value: Any) -> Any:
-    """Order-independent form of a value, so the cache key is stable across
-    processes (``repr`` of a ``set`` or unsorted ``dict`` is not)."""
-    if isinstance(value, dict):
-        items = ((k, _canonical(v)) for k, v in value.items())
-        return tuple(sorted(items, key=lambda kv: repr(kv[0])))
-    if isinstance(value, (set, frozenset)):
-        return tuple(sorted((_canonical(v) for v in value), key=repr))
-    if isinstance(value, (list, tuple)):
-        return tuple(_canonical(v) for v in value)
-    if _DEFAULT_REPR.search(repr(value)):
-        warnings.warn(
-            f"llm param {type(value).__name__!r} has no stable repr; it breaks "
-            "caching (full recompute every run).",
-            stacklevel=2,
-        )
-    return value
-
-
 def _is_secret_key(key: Any) -> bool:
     if not isinstance(key, str):
         return False
@@ -183,7 +158,7 @@ def identity(self, model: str, llm_params: Any = None) -> tuple:
         if self.schema is not None:
             elem, is_list = _element_type(self.schema)
             if hasattr(elem, "model_json_schema"):
-                schema_repr = (_canonical(elem.model_json_schema()), is_list)
+                schema_repr = (normalize_hash_value(elem.model_json_schema()), is_list)
             else:
                 schema_repr = str(self.schema)
         params = self.params
@@ -200,7 +175,7 @@ def identity(self, model: str, llm_params: Any = None) -> tuple:
             self.context_col,
             self.type,
             self.include_usage,
-            _canonical(params),
+            normalize_hash_value(params),
         )
 
     def _resolve_model(self, settings: "Settings") -> str:
diff --git a/tests/unit/lib/test_arrow.py b/tests/unit/lib/test_arrow.py
index 6977db372..8182c5fd5 100644
--- a/tests/unit/lib/test_arrow.py
+++ b/tests/unit/lib/test_arrow.py
@@ -19,6 +19,18 @@
 from datachain.lib.hf import HFClassLabel
 
 
+def test_arrow_generator_constructor_hash():
+    first_schema = dict_to_data_model("", {"value": int})
+    second_schema = dict_to_data_model("", {"value": int})
+
+    first = ArrowGenerator(output_schema=first_schema)
+    second = ArrowGenerator(output_schema=second_schema)
+    limited = ArrowGenerator(output_schema=second_schema, nrows=1)
+
+    assert first._constructor_state_hash == second._constructor_state_hash
+    assert first._constructor_state_hash != limited._constructor_state_hash
+
+
 @pytest.mark.parametrize("cache", [True, False])
 def test_arrow_generator(tmp_path, catalog, cache):
     ids = [12345, 67890, 34, 0xF0123]
diff --git a/tests/unit/lib/test_hf.py b/tests/unit/lib/test_hf.py
index 51181581e..d6b13c2b3 100644
--- a/tests/unit/lib/test_hf.py
+++ b/tests/unit/lib/test_hf.py
@@ -9,6 +9,19 @@
 )
 
 
+def test_hf_generator_constructor_hash():
+    ds = Dataset.from_dict({"value": [1]})
+    first_schema = dict_to_data_model("", {"value": int})
+    second_schema = dict_to_data_model("", {"value": int})
+
+    first = HFGenerator(ds, first_schema)
+    second = HFGenerator(ds, second_schema)
+    limited = HFGenerator(ds, second_schema, limit=1)
+
+    assert first._constructor_state_hash == second._constructor_state_hash
+    assert first._constructor_state_hash != limited._constructor_state_hash
+
+
 def test_hf():
     ds = Dataset.from_dict({"pokemon": ["bulbasaur", "squirtle"]})
     schema, norm_names = get_output_schema(ds.features)
diff --git a/tests/unit/lib/test_udf.py b/tests/unit/lib/test_udf.py
index 4e090d1cd..0ac755490 100644
--- a/tests/unit/lib/test_udf.py
+++ b/tests/unit/lib/test_udf.py
@@ -1,4 +1,7 @@
+import os
 import pickle
+import subprocess
+import sys
 
 import pytest
 from cloudpickle import dumps, loads
@@ -167,6 +170,108 @@ def process(self, x: int) -> int:
     assert udf_a.hash() == udf_b.hash()
 
 
+@pytest.mark.parametrize(
+    "args,kwargs,matches_default",
+    [
+        ((), {}, True),
+        ((3,), {}, True),
+        ((), {"limit": 3}, True),
+        ((4,), {}, False),
+    ],
+)
+def test_class_udf_captures_normalized_constructor_arguments(
+    args, kwargs, matches_default
+):
+    class Limited(Mapper):
+        def __init__(self, limit: int = 3):
+            self.limit = limit
+
+        def process(self, x: int) -> int:
+            return x + self.limit
+
+    baseline = Limited(limit=3)
+    udf = Limited(*args, **kwargs)
+
+    assert (
+        udf._constructor_state_hash == baseline._constructor_state_hash
+    ) is matches_default
+
+
+def test_class_udf_constructor_hash_survives_cloudpickle_roundtrip():
+    class Limited(Mapper):
+        def __init__(self, limit: int):
+            self.limit = limit
+
+        def process(self, x: int) -> int:
+            return x + self.limit
+
+    udf = Limited(3)
+    restored = loads(dumps(udf))
+
+    assert restored._constructor_state_hash == udf._constructor_state_hash
+
+
+def test_class_udf_hash_is_deterministic_across_processes():
+    code = """
+from datachain import Mapper
+from datachain.lib.signal_schema import SignalSchema
+from datachain.lib.udf_signature import UdfSignature
+
+class Limited(Mapper):
+    def __init__(self, limit=3, labels=frozenset({"a", "b", "c"})):
+        self.limit = limit
+        self.labels = labels
+
+    def process(self, x):
+        return x + self.limit
+
+udf = Limited()
+sign = UdfSignature(
+    udf,
+    SignalSchema({"x": int}),
+    SignalSchema({"y": int}),
+)
+print(Mapper._create(sign, sign.output_schema).hash())
+"""
+
+    hashes = {
+        subprocess.check_output(  # noqa: S603
+            [sys.executable, "-c", code],
+            env={**os.environ, "PYTHONHASHSEED": seed},
+            text=True,
+        ).strip()
+        for seed in ("1", "2", "random")
+    }
+
+    assert len(hashes) == 1
+
+
+def test_class_udf_hash_without_body_ignores_process_implementation():
+    def make_udf(add: bool):
+        class Stateful(Mapper):
+            def __init__(self, limit: int = 3):
+                self.limit = limit
+
+            if add:
+
+                def process(self, x: int) -> int:
+                    return x + self.limit
+
+            else:
+
+                def process(self, x: int) -> int:
+                    return x * self.limit
+
+        sign = get_sign(Stateful(), output="y")  # type: ignore[arg-type]
+        return Mapper._create(sign, sign.output_schema)
+
+    added = make_udf(add=True)
+    multiplied = make_udf(add=False)
+
+    assert added.hash(include_body=False) == multiplied.hash(include_body=False)
+    assert added.hash() != multiplied.hash()
+
+
 def test_udf_does_not_traverse_setup_value():
     value = {}
     value["self"] = value
diff --git a/tests/unit/test_hash_utils.py b/tests/unit/test_hash_utils.py
index 8bda8d3d1..dd76453ef 100644
--- a/tests/unit/test_hash_utils.py
+++ b/tests/unit/test_hash_utils.py
@@ -16,7 +16,46 @@
 from sqlalchemy import func as sa_func
 
 from datachain import C, func
-from datachain.hash_utils import hash_callable, hash_column_elements
+from datachain.hash_utils import (
+    hash_callable,
+    hash_column_elements,
+    normalize_hash_value,
+)
+
+
+@pytest.mark.parametrize(
+    "value,expected",
+    [
+        (
+            {"y": 2, "x": 1},
+            ("dict", (("x", 1), ("y", 2))),
+        ),
+        (
+            {"outer": {"y": 2, "x": 1}},
+            ("dict", (("outer", ("dict", (("x", 1), ("y", 2)))),)),
+        ),
+        ({"c", "b", "a"}, ("set", ("a", "b", "c"))),
+        (frozenset({2, 1}), ("frozenset", (1, 2))),
+        ([1, 2], ("list", (1, 2))),
+        ((1, 2), ("tuple", (1, 2))),
+        (None, None),
+        (True, True),
+        (1, 1),
+        (1.5, 1.5),
+        ("value", "value"),
+        (b"value", b"value"),
+    ],
+)
+def test_normalize_hash_value(value, expected):
+    assert normalize_hash_value(value) == expected
+
+
+def test_normalize_hash_value_warns_for_unstable_repr():
+    class Opaque:
+        pass
+
+    with pytest.warns(UserWarning, match="no stable repr"):
+        normalize_hash_value(Opaque())
 
 
 def double(x):
diff --git a/tests/unit/test_query_steps_hash.py b/tests/unit/test_query_steps_hash.py
index dad22e220..88242d860 100644
--- a/tests/unit/test_query_steps_hash.py
+++ b/tests/unit/test_query_steps_hash.py
@@ -322,7 +322,7 @@ def test_subtract_hash(test_session, numbers_dataset, on):
             DoubleMapper(),
             ["x"],
             {"double": int},
-            "a91d1f4093958aff06588e39b69892eb108d22c452717d15740b271bfd3a6157",
+            "7c901f584e52f41ac22a5fed332da2c3093d7d03f0fc16a780a9ad11eb456379",
         ),
     ],
 )
@@ -362,7 +362,7 @@ def test_udf_mapper_hash(
             TripleGenerator(),
             ["x"],
             {"triple": int},
-            "b158fe7fb7396352ba1c4e47a92f70b3ec261228eb0968292340fb75ab60eea1",
+            "208bd4e553088f51983cadac3f645893bc44ffec7ccd625e28d7dfa0ea000170",
         ),
     ],
 )

From c4916e5c1cbe59e86f04fcd96a9f4f50b3c13721 Mon Sep 17 00:00:00 2001
From: ilongin <ivan.longin1@gmail.com>
Date: Wed, 19 Aug 2026 14:58:20 +0200
Subject: [PATCH 04/16] refactoring

---
 src/datachain/lib/udf.py   | 13 ++++---------
 tests/unit/lib/test_llm.py |  7 -------
 2 files changed, 4 insertions(+), 16 deletions(-)

diff --git a/src/datachain/lib/udf.py b/src/datachain/lib/udf.py
index e3c056137..a7dcebc3b 100644
--- a/src/datachain/lib/udf.py
+++ b/src/datachain/lib/udf.py
@@ -232,22 +232,17 @@ def process(self, file) -> list[float]:
 
     def __new__(cls, *args, **kwargs):
         instance = super().__new__(cls)
-        signature = inspect.signature(cls.__init__)
-
         try:
-            bound = signature.bind(instance, *args, **kwargs)
+            # 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 = dict(bound.arguments)
-        if signature.parameters:
-            self_name = next(iter(signature.parameters))
-            arguments.pop(self_name, None)
-
-        arguments = cls._constructor_hash_args(arguments)
+        arguments = cls._constructor_hash_args(dict(bound.arguments))
         instance._constructor_state_hash = hash_value(arguments)
         return instance
 
diff --git a/tests/unit/lib/test_llm.py b/tests/unit/lib/test_llm.py
index ae745b081..b2919ba05 100644
--- a/tests/unit/lib/test_llm.py
+++ b/tests/unit/lib/test_llm.py
@@ -511,13 +511,6 @@ def test_identity_stable_across_param_dict_order():
     assert a == b
 
 
-def test_canonical_orders_sets_and_dicts():
-    from datachain.llm.spec import _canonical
-
-    assert _canonical({"a", "b", "c"}) == _canonical({"c", "b", "a"})
-    assert _canonical({"x": 1, "y": 2}) == _canonical({"y": 2, "x": 1})
-
-
 def test_param_clobber_is_blocked(fake_llm):
     bind(llm.complete("t", "real prompt", temperature=0.0), llm="real/model")("hi")
     call = fake_llm.calls[-1]

From 08fc4ba160ac10642170b8d6e9ca4371d2d9baea Mon Sep 17 00:00:00 2001
From: ilongin <ivan.longin1@gmail.com>
Date: Thu, 20 Aug 2026 16:06:25 +0200
Subject: [PATCH 05/16] fixing issue with repr

---
 src/datachain/hash_utils.py   | 35 ++++++++++--------------
 src/datachain/lib/arrow.py    | 34 +++++++++++++++++++++++
 src/datachain/lib/hf.py       | 14 ++++++++++
 src/datachain/lib/udf.py      | 14 +++++++++-
 src/datachain/llm/spec.py     | 19 +++++++++++--
 tests/unit/lib/test_arrow.py  | 19 ++++++++++---
 tests/unit/lib/test_hf.py     | 16 +++++++----
 tests/unit/lib/test_llm.py    | 20 +++++++-------
 tests/unit/lib/test_udf.py    | 24 +++++++++++++++++
 tests/unit/test_hash_utils.py | 51 +++++++++++++++++++++++++----------
 10 files changed, 191 insertions(+), 55 deletions(-)

diff --git a/src/datachain/hash_utils.py b/src/datachain/hash_utils.py
index d1c71e503..1aa561626 100644
--- a/src/datachain/hash_utils.py
+++ b/src/datachain/hash_utils.py
@@ -1,9 +1,7 @@
 import hashlib
 import inspect
 import logging
-import re
 import textwrap
-import warnings
 from collections.abc import Sequence
 from typing import TypeAlias, TypeVar
 from uuid import uuid4
@@ -18,40 +16,35 @@
 ColumnLike: TypeAlias = str | T
 
 
-# A value with no custom repr shows its memory address (``<Cls object at 0x...>``
-# for plain objects, ``<function name at 0x...>`` for functions and methods),
-# which changes each run and would make the cache key unstable.
-_DEFAULT_REPR = re.compile(r" at 0x[0-9a-fA-F]+>")
+def normalize_hash_value(value):  # noqa: PLR0911
+    """Return a complete, deterministic representation for stable cache keys."""
+    value_type = type(value)
 
-
-def normalize_hash_value(value):
-    """Order-independent form of a value for stable cache keys."""
-    if isinstance(value, dict):
+    if value is None:
+        return ("none",)
+    if value_type in (bool, int, float, str, bytes):
+        return (value_type.__name__, value)
+    if value_type is dict:
         items = (
             (normalize_hash_value(k), normalize_hash_value(v)) for k, v in value.items()
         )
         return ("dict", tuple(sorted(items, key=repr)))
-    if isinstance(value, set):
+    if value_type is set:
         return (
             "set",
             tuple(sorted(map(normalize_hash_value, value), key=repr)),
         )
-    if isinstance(value, frozenset):
+    if value_type is frozenset:
         return (
             "frozenset",
             tuple(sorted(map(normalize_hash_value, value), key=repr)),
         )
-    if isinstance(value, list):
+    if value_type is list:
         return ("list", tuple(map(normalize_hash_value, value)))
-    if isinstance(value, tuple):
+    if value_type is tuple:
         return ("tuple", tuple(map(normalize_hash_value, value)))
-    if _DEFAULT_REPR.search(repr(value)):
-        warnings.warn(
-            f"value of type {type(value).__name__!r} has no stable repr; it "
-            "breaks caching (full recompute every run).",
-            stacklevel=2,
-        )
-    return value
+
+    raise TypeError(f"value of type {value_type.__name__!r} cannot be hashed safely")
 
 
 def hash_value(value) -> str:
diff --git a/src/datachain/lib/arrow.py b/src/datachain/lib/arrow.py
index e15e0cbdf..c188c92a1 100644
--- a/src/datachain/lib/arrow.py
+++ b/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
@@ -9,6 +10,7 @@
 
 from datachain import json
 from datachain.fs.reference import ReferenceFileSystem
+from datachain.hash_utils import hash_callable
 from datachain.lib.convert.flatten import classify_field, iter_flat_columns
 from datachain.lib.data_model import (
     NULLABLE_SCALARS,
@@ -33,6 +35,28 @@
 DATACHAIN_SIGNAL_SCHEMA_PARQUET_KEY = b"DataChain SignalSchema"
 
 
+def _parse_options_hash_args(options: ParseOptions) -> dict[str, Any]:
+    handler = options.invalid_row_handler
+    return {
+        "delimiter": options.delimiter,
+        "double_quote": options.double_quote,
+        "escape_char": options.escape_char,
+        "ignore_empty_lines": options.ignore_empty_lines,
+        "invalid_row_handler": hash_callable(handler) if handler else None,
+        "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 (
@@ -83,6 +107,16 @@ def _constructor_hash_args(cls, arguments):
         # deterministic across runs.
         arguments = arguments.copy()
         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):
diff --git a/src/datachain/lib/hf.py b/src/datachain/lib/hf.py
index 0cdf55395..8f1f55f6d 100644
--- a/src/datachain/lib/hf.py
+++ b/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
@@ -101,6 +114,7 @@ def _constructor_hash_args(cls, arguments):
         # deterministic across runs.
         arguments = arguments.copy()
         arguments.pop("output_schema", None)
+        arguments["ds"] = _dataset_hash_args(arguments["ds"])
         return arguments
 
     def setup(self):
diff --git a/src/datachain/lib/udf.py b/src/datachain/lib/udf.py
index a7dcebc3b..53aaf976f 100644
--- a/src/datachain/lib/udf.py
+++ b/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
@@ -175,6 +176,17 @@ def prefetch(self) -> int:
         return self.inner.prefetch
 
 
+def _hash_constructor_args(arguments: dict[str, Any]) -> str:
+    try:
+        return hash_value(arguments)
+    except TypeError as exc:
+        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.
 
@@ -243,7 +255,7 @@ def __new__(cls, *args, **kwargs):
 
         bound.apply_defaults()
         arguments = cls._constructor_hash_args(dict(bound.arguments))
-        instance._constructor_state_hash = hash_value(arguments)
+        instance._constructor_state_hash = _hash_constructor_args(arguments)
         return instance
 
     @classmethod
diff --git a/src/datachain/llm/spec.py b/src/datachain/llm/spec.py
index 99d631206..17c5708d7 100644
--- a/src/datachain/llm/spec.py
+++ b/src/datachain/llm/spec.py
@@ -1,6 +1,8 @@
+import logging
 from collections.abc import Callable, Iterator
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any, Literal, get_args, get_origin
+from uuid import uuid4
 
 from pydantic import BaseModel
 
@@ -13,6 +15,8 @@
 if TYPE_CHECKING:
     from datachain.lib.settings import Settings
 
+logger = logging.getLogger("datachain")
+
 
 class LLMConfigError(engine.LLMError):
     """Raised when no model can be resolved for a `datachain.llm` operation."""
@@ -46,6 +50,14 @@ def _without_secrets(value: Any) -> Any:
     return value
 
 
+def _normalize_identity_value(value: Any) -> Any:
+    try:
+        return normalize_hash_value(value)
+    except TypeError as exc:
+        logger.warning("%s; cache reuse for this LLM operation is disabled", exc)
+        return ("unsupported", uuid4().hex)
+
+
 @dataclass
 class LLMSpec(BoundSpec):
     """A configured `datachain.llm` operation, used inside `.map()` / `.gen()`.
@@ -158,7 +170,10 @@ def identity(self, model: str, llm_params: Any = None) -> tuple:
         if self.schema is not None:
             elem, is_list = _element_type(self.schema)
             if hasattr(elem, "model_json_schema"):
-                schema_repr = (normalize_hash_value(elem.model_json_schema()), is_list)
+                schema_repr = (
+                    _normalize_identity_value(elem.model_json_schema()),
+                    is_list,
+                )
             else:
                 schema_repr = str(self.schema)
         params = self.params
@@ -175,7 +190,7 @@ def identity(self, model: str, llm_params: Any = None) -> tuple:
             self.context_col,
             self.type,
             self.include_usage,
-            normalize_hash_value(params),
+            _normalize_identity_value(params),
         )
 
     def _resolve_model(self, settings: "Settings") -> str:
diff --git a/tests/unit/lib/test_arrow.py b/tests/unit/lib/test_arrow.py
index 8182c5fd5..e0a65298f 100644
--- a/tests/unit/lib/test_arrow.py
+++ b/tests/unit/lib/test_arrow.py
@@ -6,6 +6,8 @@
 import pyarrow.parquet as pq
 import pytest
 from datasets import Dataset
+from pyarrow.csv import ParseOptions
+from pyarrow.dataset import CsvFileFormat
 
 import datachain as dc
 from datachain.lib.arrow import (
@@ -23,9 +25,20 @@ def test_arrow_generator_constructor_hash():
     first_schema = dict_to_data_model("", {"value": int})
     second_schema = dict_to_data_model("", {"value": int})
 
-    first = ArrowGenerator(output_schema=first_schema)
-    second = ArrowGenerator(output_schema=second_schema)
-    limited = ArrowGenerator(output_schema=second_schema, nrows=1)
+    def make_generator(output_schema, nrows=None):
+        input_schema = pa.schema({"value": pa.int64()})
+        parse_options = ParseOptions(delimiter=";")
+        return ArrowGenerator(
+            input_schema=input_schema,
+            output_schema=output_schema,
+            nrows=nrows,
+            parse_options=parse_options,
+            format=CsvFileFormat(parse_options=parse_options),
+        )
+
+    first = make_generator(first_schema)
+    second = make_generator(second_schema)
+    limited = make_generator(second_schema, nrows=1)
 
     assert first._constructor_state_hash == second._constructor_state_hash
     assert first._constructor_state_hash != limited._constructor_state_hash
diff --git a/tests/unit/lib/test_hf.py b/tests/unit/lib/test_hf.py
index d6b13c2b3..8516d01a7 100644
--- a/tests/unit/lib/test_hf.py
+++ b/tests/unit/lib/test_hf.py
@@ -1,3 +1,4 @@
+import pytest
 from datasets import Array2D, Dataset, DatasetDict, Sequence, Value
 
 from datachain.lib.data_model import dict_to_data_model
@@ -9,14 +10,19 @@
 )
 
 
-def test_hf_generator_constructor_hash():
-    ds = Dataset.from_dict({"value": [1]})
+@pytest.mark.parametrize("as_dict", [False, True])
+def test_hf_generator_constructor_hash(as_dict):
+    first_ds = Dataset.from_dict({"value": [1]})
+    second_ds = Dataset.from_dict({"value": [1]})
+    if as_dict:
+        first_ds = DatasetDict({"train": first_ds})
+        second_ds = DatasetDict({"train": second_ds})
     first_schema = dict_to_data_model("", {"value": int})
     second_schema = dict_to_data_model("", {"value": int})
 
-    first = HFGenerator(ds, first_schema)
-    second = HFGenerator(ds, second_schema)
-    limited = HFGenerator(ds, second_schema, limit=1)
+    first = HFGenerator(first_ds, first_schema)
+    second = HFGenerator(second_ds, second_schema)
+    limited = HFGenerator(second_ds, second_schema, limit=1)
 
     assert first._constructor_state_hash == second._constructor_state_hash
     assert first._constructor_state_hash != limited._constructor_state_hash
diff --git a/tests/unit/lib/test_llm.py b/tests/unit/lib/test_llm.py
index 7892e1adc..52de2c4d8 100644
--- a/tests/unit/lib/test_llm.py
+++ b/tests/unit/lib/test_llm.py
@@ -726,20 +726,22 @@ def test_callable_llm_params_not_in_identity():
     assert spec.identity("m", lambda: {"k": "v"}) == spec.identity("m")
 
 
-def test_opaque_param_value_warns_about_unstable_cache_key():
+def test_opaque_param_value_disables_cache_reuse():
     class Opaque:
-        pass
+        def __repr__(self):
+            return "Opaque()"
 
-    with pytest.warns(UserWarning, match="no stable repr"):
-        llm.complete("t", client=Opaque()).identity("m")
+    first = llm.complete("t", client=Opaque()).identity("m")
+    second = llm.complete("t", client=Opaque()).identity("m")
 
+    assert first != second
 
-def test_stable_param_values_do_not_warn():
-    import warnings
 
-    with warnings.catch_warnings():
-        warnings.simplefilter("error")
-        llm.complete("t", temperature=0.0, opt={"a": 1}).identity("m")
+def test_stable_param_values_have_stable_identity():
+    first = llm.complete("t", temperature=0.0, opt={"a": 1}).identity("m")
+    second = llm.complete("t", temperature=0.0, opt={"a": 1}).identity("m")
+
+    assert first == second
 
 
 def test_secret_params_not_in_identity():
diff --git a/tests/unit/lib/test_udf.py b/tests/unit/lib/test_udf.py
index 0ac755490..166a341dd 100644
--- a/tests/unit/lib/test_udf.py
+++ b/tests/unit/lib/test_udf.py
@@ -170,6 +170,30 @@ def process(self, x: int) -> int:
     assert udf_a.hash() == udf_b.hash()
 
 
+def test_class_udf_unsupported_constructor_value_disables_cache_reuse(caplog):
+    class Opaque:
+        def __repr__(self):
+            return "Opaque()"
+
+    class Configured(Mapper):
+        def __init__(self, config: Opaque):
+            self.config = config
+
+        def process(self, x: int) -> int:
+            return x
+
+    first = Configured(Opaque())
+    second = Configured(Opaque())
+    sign_a = get_sign(first, output="y")
+    sign_b = get_sign(second, output="y")
+    udf_a = Mapper._create(sign_a, sign_a.output_schema)
+    udf_b = Mapper._create(sign_b, sign_b.output_schema)
+
+    assert udf_a.hash() == udf_a.hash()
+    assert udf_a.hash() != udf_b.hash()
+    assert "cache reuse across UDF instances is disabled" in caplog.text
+
+
 @pytest.mark.parametrize(
     "args,kwargs,matches_default",
     [
diff --git a/tests/unit/test_hash_utils.py b/tests/unit/test_hash_utils.py
index dd76453ef..97922bd84 100644
--- a/tests/unit/test_hash_utils.py
+++ b/tests/unit/test_hash_utils.py
@@ -28,33 +28,56 @@
     [
         (
             {"y": 2, "x": 1},
-            ("dict", (("x", 1), ("y", 2))),
+            (
+                "dict",
+                (
+                    (("str", "x"), ("int", 1)),
+                    (("str", "y"), ("int", 2)),
+                ),
+            ),
         ),
         (
             {"outer": {"y": 2, "x": 1}},
-            ("dict", (("outer", ("dict", (("x", 1), ("y", 2)))),)),
+            (
+                "dict",
+                (
+                    (
+                        ("str", "outer"),
+                        (
+                            "dict",
+                            (
+                                (("str", "x"), ("int", 1)),
+                                (("str", "y"), ("int", 2)),
+                            ),
+                        ),
+                    ),
+                ),
+            ),
         ),
-        ({"c", "b", "a"}, ("set", ("a", "b", "c"))),
-        (frozenset({2, 1}), ("frozenset", (1, 2))),
-        ([1, 2], ("list", (1, 2))),
-        ((1, 2), ("tuple", (1, 2))),
-        (None, None),
-        (True, True),
-        (1, 1),
-        (1.5, 1.5),
-        ("value", "value"),
-        (b"value", b"value"),
+        (
+            {"c", "b", "a"},
+            ("set", (("str", "a"), ("str", "b"), ("str", "c"))),
+        ),
+        (frozenset({2, 1}), ("frozenset", (("int", 1), ("int", 2)))),
+        ([1, 2], ("list", (("int", 1), ("int", 2)))),
+        ((1, 2), ("tuple", (("int", 1), ("int", 2)))),
+        (None, ("none",)),
+        (True, ("bool", True)),
+        (1, ("int", 1)),
+        (1.5, ("float", 1.5)),
+        ("value", ("str", "value")),
+        (b"value", ("bytes", b"value")),
     ],
 )
 def test_normalize_hash_value(value, expected):
     assert normalize_hash_value(value) == expected
 
 
-def test_normalize_hash_value_warns_for_unstable_repr():
+def test_normalize_hash_value_rejects_unsupported_value():
     class Opaque:
         pass
 
-    with pytest.warns(UserWarning, match="no stable repr"):
+    with pytest.raises(TypeError, match="cannot be hashed safely"):
         normalize_hash_value(Opaque())
 
 

From b686bc66894f5394c04f9094914be634a79df5df Mon Sep 17 00:00:00 2001
From: ilongin <ivan.longin1@gmail.com>
Date: Wed, 26 Aug 2026 15:51:37 +0200
Subject: [PATCH 06/16] fixing arrow generator

---
 src/datachain/lib/arrow.py   |  4 +---
 tests/unit/lib/test_arrow.py | 18 ++++++++++++++++++
 2 files changed, 19 insertions(+), 3 deletions(-)

diff --git a/src/datachain/lib/arrow.py b/src/datachain/lib/arrow.py
index c188c92a1..2aee52cec 100644
--- a/src/datachain/lib/arrow.py
+++ b/src/datachain/lib/arrow.py
@@ -10,7 +10,6 @@
 
 from datachain import json
 from datachain.fs.reference import ReferenceFileSystem
-from datachain.hash_utils import hash_callable
 from datachain.lib.convert.flatten import classify_field, iter_flat_columns
 from datachain.lib.data_model import (
     NULLABLE_SCALARS,
@@ -36,13 +35,12 @@
 
 
 def _parse_options_hash_args(options: ParseOptions) -> dict[str, Any]:
-    handler = options.invalid_row_handler
     return {
         "delimiter": options.delimiter,
         "double_quote": options.double_quote,
         "escape_char": options.escape_char,
         "ignore_empty_lines": options.ignore_empty_lines,
-        "invalid_row_handler": hash_callable(handler) if handler else None,
+        "invalid_row_handler": options.invalid_row_handler,
         "newlines_in_values": options.newlines_in_values,
         "quote_char": options.quote_char,
     }
diff --git a/tests/unit/lib/test_arrow.py b/tests/unit/lib/test_arrow.py
index e0a65298f..582ed9238 100644
--- a/tests/unit/lib/test_arrow.py
+++ b/tests/unit/lib/test_arrow.py
@@ -44,6 +44,24 @@ def make_generator(output_schema, nrows=None):
     assert first._constructor_state_hash != limited._constructor_state_hash
 
 
+def test_arrow_generator_constructor_hash_with_closure_handler():
+    def make_handler(prefix):
+        def handler(row):
+            return "skip" if row.text.startswith(prefix) else "error"
+
+        return handler
+
+    def make_generator(prefix):
+        return ArrowGenerator(
+            parse_options=ParseOptions(invalid_row_handler=make_handler(prefix))
+        )
+
+    comments = make_generator("#")
+    metadata = make_generator("!")
+
+    assert comments._constructor_state_hash != metadata._constructor_state_hash
+
+
 @pytest.mark.parametrize("cache", [True, False])
 def test_arrow_generator(tmp_path, catalog, cache):
     ids = [12345, 67890, 34, 0xF0123]

From 7c0e5de8a2c95ec2bfffe937bf9fd2cfd0abd0fc Mon Sep 17 00:00:00 2001
From: ilongin <ivan.longin1@gmail.com>
Date: Wed, 26 Aug 2026 16:21:16 +0200
Subject: [PATCH 07/16] fixing llm hashing

---
 src/datachain/llm/spec.py  | 18 +++++++++++++-----
 tests/unit/lib/test_llm.py |  9 +++++++++
 2 files changed, 22 insertions(+), 5 deletions(-)

diff --git a/src/datachain/llm/spec.py b/src/datachain/llm/spec.py
index 39f323a44..fb2d9fbed 100644
--- a/src/datachain/llm/spec.py
+++ b/src/datachain/llm/spec.py
@@ -50,12 +50,12 @@ def _without_secrets(value: Any) -> Any:
     return value
 
 
-def _normalize_identity_value(value: Any) -> Any:
+def _normalize_identity_value(value: Any, fallback: str) -> Any:
     try:
         return normalize_hash_value(value)
     except TypeError as exc:
         logger.warning("%s; cache reuse for this LLM operation is disabled", exc)
-        return ("unsupported", uuid4().hex)
+        return ("unsupported", fallback)
 
 
 @dataclass
@@ -78,6 +78,12 @@ class LLMSpec(BoundSpec):
     fallback: str | list[str] | None = None
     include_usage: bool = False
     params: dict[str, Any] = field(default_factory=dict)
+    _identity_fallback: str = field(
+        default_factory=lambda: uuid4().hex,
+        init=False,
+        repr=False,
+        compare=False,
+    )
 
     def __post_init__(self) -> None:
         if self.schema is not None:
@@ -153,7 +159,7 @@ def return_annotation(self, to_many: bool = False) -> Any:
         return tuple[out, Usage] if self.include_usage else out  # type: ignore[valid-type]
 
     def identity(self, model: str, llm_params: Any = None) -> tuple:
-        """Cache key baked into the UDF hash; changes iff an output-affecting
+        """Cache key baked into the UDF hash; changes if an output-affecting
         input (model, prompt, schema, params, llm_params, ...) changes.
 
         The schema is keyed by its JSON schema (fields, types, constraints, name);
@@ -171,7 +177,9 @@ def identity(self, model: str, llm_params: Any = None) -> tuple:
             elem, is_list = _element_type(self.schema)
             if hasattr(elem, "model_json_schema"):
                 schema_repr = (
-                    _normalize_identity_value(elem.model_json_schema()),
+                    _normalize_identity_value(
+                        elem.model_json_schema(), self._identity_fallback
+                    ),
                     is_list,
                 )
             else:
@@ -190,7 +198,7 @@ def identity(self, model: str, llm_params: Any = None) -> tuple:
             self.context_col,
             self.type,
             self.include_usage,
-            _normalize_identity_value(params),
+            _normalize_identity_value(params, self._identity_fallback),
         )
 
     def _resolve_model(self, settings: "Settings") -> str:
diff --git a/tests/unit/lib/test_llm.py b/tests/unit/lib/test_llm.py
index a8549e250..ca2b78419 100644
--- a/tests/unit/lib/test_llm.py
+++ b/tests/unit/lib/test_llm.py
@@ -736,6 +736,15 @@ def __repr__(self):
     assert first != second
 
 
+def test_opaque_param_value_has_stable_identity_within_operation():
+    class Opaque:
+        pass
+
+    spec = llm.complete("t", client=Opaque())
+
+    assert spec.identity("m") == spec.identity("m")
+
+
 def test_stable_param_values_have_stable_identity():
     first = llm.complete("t", temperature=0.0, opt={"a": 1}).identity("m")
     second = llm.complete("t", temperature=0.0, opt={"a": 1}).identity("m")

From f10cfa047f7673d86cbda5db4055665a62052b70 Mon Sep 17 00:00:00 2001
From: ilongin <ivan.longin1@gmail.com>
Date: Thu, 27 Aug 2026 14:18:11 +0200
Subject: [PATCH 08/16] refactoring

---
 docs/guide/checkpoints.md   |   1 +
 docs/guide/python-engine.md |  27 +++++++++
 docs/references/llm.md      |   3 +-
 src/datachain/lib/udf.py    |  27 ++++++++-
 tests/unit/lib/test_udf.py  | 114 ++++++++++++++++++++++++++++++++++++
 5 files changed, 170 insertions(+), 2 deletions(-)

diff --git a/docs/guide/checkpoints.md b/docs/guide/checkpoints.md
index b60c7cfef..6d153a165 100644
--- a/docs/guide/checkpoints.md
+++ b/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.
diff --git a/docs/guide/python-engine.md b/docs/guide/python-engine.md
index 561131ae6..bc28cdf90 100644
--- a/docs/guide/python-engine.md
+++ b/docs/guide/python-engine.md
@@ -151,6 +151,33 @@ 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
+from datachain.lib.udf import Mapper
+
+class Tokenize(Mapper):
+    def __init__(self, tokenizer, tokenizer_version: str):
+        self.tokenizer = tokenizer
+        self.tokenizer_version = tokenizer_version
+
+    def state_hash(self) -> str:
+        return hashlib.sha256(self.tokenizer_version.encode()).hexdigest()
+
+    def process(self, text: str) -> list[str]:
+        return self.tokenizer(text)
+```
+
+`state_hash()` must return a SHA-256 hexadecimal string covering all instance state
+that affects output. It replaces automatic constructor hashing; UDF code and schemas
+are still included. An incomplete hash can reuse an incorrect cached result.
+
 Use class-based operations sparingly; `.setup()` covers most cases.
 
 ## Execution and Scale
diff --git a/docs/references/llm.md b/docs/references/llm.md
index a52a1087e..ca9537206 100644
--- a/docs/references/llm.md
+++ b/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
 
diff --git a/src/datachain/lib/udf.py b/src/datachain/lib/udf.py
index 9a4311c12..71b0678af 100644
--- a/src/datachain/lib/udf.py
+++ b/src/datachain/lib/udf.py
@@ -296,6 +296,10 @@ def process(self, file) -> list[float]:
 
     def __new__(cls, *args, **kwargs):
         instance = super().__new__(cls)
+        if cls.state_hash is not UDFBase.state_hash:
+            # A public state_hash() override owns instance identity, so avoid
+            # normalizing constructor values or warning about opaque arguments.
+            return instance
         try:
             # bound-method signature so `self` is already stripped; correct even
             # when __init__ uses *args instead of a named `self` parameter.
@@ -315,6 +319,16 @@ 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.
+        """
+        return self._constructor_state_hash
+
     def __init__(self):
         self.params: SignalSchema | None = None
         self.output = None
@@ -343,7 +357,18 @@ def hash(self, include_body: bool = True) -> str:
         # 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:
-            parts.append(self._constructor_state_hash)
+            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:
+                bytes.fromhex(state_hash)
+            except ValueError as exc:
+                raise ValueError(
+                    "state_hash() must return a SHA-256 hexadecimal string"
+                ) from exc
+            parts.append(state_hash)
 
         return hashlib.sha256(
             b"".join([bytes.fromhex(part) for part in parts])
diff --git a/tests/unit/lib/test_udf.py b/tests/unit/lib/test_udf.py
index 63de2ca23..fb212bf55 100644
--- a/tests/unit/lib/test_udf.py
+++ b/tests/unit/lib/test_udf.py
@@ -11,6 +11,7 @@
 import datachain as dc
 from datachain import Mapper
 from datachain.dataset import RowDict
+from datachain.hash_utils import hash_value
 from datachain.lib.file import File
 from datachain.lib.signal_schema import SignalSchema
 from datachain.lib.udf import JsonSerializationError, UDFBase, UdfError, UdfRunError
@@ -19,6 +20,31 @@
 from .test_udf_signature import get_sign
 
 
+class _OpaqueConstructorValue:
+    pass
+
+
+class _HashableConstructorValue:
+    def __hash__(self):
+        return 1
+
+
+class _CallableConstructorValue:
+    def __call__(self, value):
+        return value
+
+
+def _constructor_function(value):
+    return value
+
+
+def _make_constructor_closure(captured):
+    def closure(value):
+        return value, captured
+
+    return closure
+
+
 def test_udf_error():
     orig_err = UdfError("test error")
     for err in (orig_err, loads(dumps(orig_err))):
@@ -194,6 +220,94 @@ def process(self, x: int) -> int:
     assert "cache reuse across UDF instances is disabled" in caplog.text
 
 
+@pytest.mark.parametrize(
+    "config",
+    [
+        pytest.param(_OpaqueConstructorValue(), id="custom-object"),
+        pytest.param(_HashableConstructorValue(), id="object-with-hash"),
+        pytest.param(_constructor_function, id="function"),
+        pytest.param(lambda value: value, id="lambda"),
+        pytest.param(_make_constructor_closure("captured"), id="closure"),
+        pytest.param(_CallableConstructorValue(), id="callable-object"),
+        pytest.param(
+            {"options": [{"client": _OpaqueConstructorValue()}]},
+            id="nested-custom-object",
+        ),
+    ],
+)
+def test_class_udf_unsupported_constructor_values_do_not_reuse_cache(config):
+    class Configured(Mapper):
+        def __init__(self, value):
+            self.value = value
+
+        def process(self, x: int) -> int:
+            return x
+
+    first = Configured(config)
+    second = Configured(config)
+    sign_a = get_sign(first, output="y")
+    sign_b = get_sign(second, output="y")
+    udf_a = Mapper._create(sign_a, sign_a.output_schema)
+    udf_b = Mapper._create(sign_b, sign_b.output_schema)
+
+    assert udf_a.hash() == udf_a.hash()
+    assert udf_a.hash() != udf_b.hash()
+
+
+@pytest.mark.parametrize(
+    "first_key,second_key,matches",
+    [
+        ("tokenizer-v1", "tokenizer-v1", True),
+        ("tokenizer-v1", "tokenizer-v2", False),
+    ],
+)
+def test_class_udf_state_hash_overrides_opaque_constructor_fallback(
+    first_key, second_key, matches, caplog
+):
+    class Opaque:
+        pass
+
+    class Configured(Mapper):
+        def __init__(self, config: Opaque, cache_key: str):
+            self.config = config
+            self.cache_key = cache_key
+
+        def state_hash(self) -> str:
+            return hash_value(self.cache_key)
+
+        def process(self, x: int) -> int:
+            return x
+
+    first = Configured(Opaque(), first_key)
+    second = Configured(Opaque(), second_key)
+    sign_a = get_sign(first, output="y")
+    sign_b = get_sign(second, output="y")
+    udf_a = Mapper._create(sign_a, sign_a.output_schema)
+    udf_b = Mapper._create(sign_b, sign_b.output_schema)
+
+    assert (udf_a.hash() == udf_b.hash()) is matches
+    assert "cache reuse across UDF instances is disabled" not in caplog.text
+
+
+@pytest.mark.parametrize("invalid_hash", ["tokenizer-v1", None])
+def test_class_udf_state_hash_rejects_invalid_hash(invalid_hash):
+    class Configured(Mapper):
+        def state_hash(self) -> str:
+            return invalid_hash
+
+        def process(self, x: int) -> int:
+            return x
+
+    udf = Configured()
+    sign = get_sign(udf, output="y")
+    udf = Mapper._create(sign, sign.output_schema)
+
+    with pytest.raises(
+        ValueError, match=r"state_hash\(\) must return a SHA-256 hexadecimal string"
+    ):
+        udf.hash()
+
+
 @pytest.mark.parametrize(
     "args,kwargs,matches_default",
     [

From 0e7e758ab8bb860fcfc23af9c169714a237cf08b Mon Sep 17 00:00:00 2001
From: ilongin <ivan.longin1@gmail.com>
Date: Thu, 27 Aug 2026 14:26:26 +0200
Subject: [PATCH 09/16] fix floating point hash

---
 src/datachain/hash_utils.py   |  5 ++++-
 tests/unit/test_hash_utils.py | 19 ++++++++++++++++++-
 2 files changed, 22 insertions(+), 2 deletions(-)

diff --git a/src/datachain/hash_utils.py b/src/datachain/hash_utils.py
index 1aa561626..20a06667f 100644
--- a/src/datachain/hash_utils.py
+++ b/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
@@ -22,7 +23,9 @@ def normalize_hash_value(value):  # noqa: PLR0911
 
     if value is None:
         return ("none",)
-    if value_type in (bool, int, float, str, bytes):
+    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 = (
diff --git a/tests/unit/test_hash_utils.py b/tests/unit/test_hash_utils.py
index 97922bd84..44839f221 100644
--- a/tests/unit/test_hash_utils.py
+++ b/tests/unit/test_hash_utils.py
@@ -1,3 +1,5 @@
+import struct
+
 import pytest
 from sqlalchemy import (
     Float,
@@ -19,6 +21,7 @@
 from datachain.hash_utils import (
     hash_callable,
     hash_column_elements,
+    hash_value,
     normalize_hash_value,
 )
 
@@ -64,7 +67,7 @@
         (None, ("none",)),
         (True, ("bool", True)),
         (1, ("int", 1)),
-        (1.5, ("float", 1.5)),
+        (1.5, ("float", struct.pack("!d", 1.5))),
         ("value", ("str", "value")),
         (b"value", ("bytes", b"value")),
     ],
@@ -81,6 +84,20 @@ class Opaque:
         normalize_hash_value(Opaque())
 
 
+@pytest.mark.parametrize(
+    "first_bits,second_bits",
+    [
+        ("7ff8000000000001", "7ff8000000000002"),
+        ("7ff8000000000001", "fff8000000000001"),
+    ],
+)
+def test_hash_value_distinguishes_float_bit_patterns(first_bits, second_bits):
+    first = struct.unpack("!d", bytes.fromhex(first_bits))[0]
+    second = struct.unpack("!d", bytes.fromhex(second_bits))[0]
+
+    assert hash_value(first) != hash_value(second)
+
+
 def double(x):
     return x * 2
 

From 461027f9765bffb3fbe5cfb418b94fa28ce72704 Mon Sep 17 00:00:00 2001
From: ilongin <ivan.longin1@gmail.com>
Date: Thu, 27 Aug 2026 14:55:01 +0200
Subject: [PATCH 10/16] refactoring

---
 src/datachain/lib/udf.py   |  6 +++++-
 tests/func/test_udf.py     | 29 +++++++++++++++--------------
 tests/unit/lib/test_udf.py |  5 ++++-
 3 files changed, 24 insertions(+), 16 deletions(-)

diff --git a/src/datachain/lib/udf.py b/src/datachain/lib/udf.py
index 71b0678af..291471c4d 100644
--- a/src/datachain/lib/udf.py
+++ b/src/datachain/lib/udf.py
@@ -363,11 +363,15 @@ def hash(self, include_body: bool = True) -> str:
                     "state_hash() must return a SHA-256 hexadecimal string"
                 )
             try:
-                bytes.fromhex(state_hash)
+                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(
diff --git a/tests/func/test_udf.py b/tests/func/test_udf.py
index f5e819bb7..ad4e2f87c 100644
--- a/tests/func/test_udf.py
+++ b/tests/func/test_udf.py
@@ -275,22 +275,23 @@ def process(self, key, value):
         value=[1, 2, 3, 4, 5, 1, 2, 3, 4, 5],
         session=test_session,
     )
-    rows_zero = sorted(
-        src.agg(
-            CountAbove(0),
-            partition_by="key",
-            params=["key", "value"],
-            output={"o": Out},
-        ).to_list("o.key", "o.n")
+    chain_zero = src.agg(
+        CountAbove(0),
+        partition_by="key",
+        params=["key", "value"],
+        output={"o": Out},
     )
-    rows_three = sorted(
-        src.agg(
-            CountAbove(3),
-            partition_by="key",
-            params=["key", "value"],
-            output={"o": Out},
-        ).to_list("o.key", "o.n")
+    chain_three = src.agg(
+        CountAbove(3),
+        partition_by="key",
+        params=["key", "value"],
+        output={"o": Out},
     )
+
+    assert chain_zero._query.hash() != chain_three._query.hash()
+
+    rows_zero = sorted(chain_zero.to_list("o.key", "o.n"))
+    rows_three = sorted(chain_three.to_list("o.key", "o.n"))
     assert rows_zero == [(1, 5), (2, 5)]
     assert rows_three == [(1, 2), (2, 2)]
 
diff --git a/tests/unit/lib/test_udf.py b/tests/unit/lib/test_udf.py
index fb212bf55..796c8b105 100644
--- a/tests/unit/lib/test_udf.py
+++ b/tests/unit/lib/test_udf.py
@@ -289,7 +289,10 @@ def process(self, x: int) -> int:
     assert "cache reuse across UDF instances is disabled" not in caplog.text
 
 
-@pytest.mark.parametrize("invalid_hash", ["tokenizer-v1", None])
+@pytest.mark.parametrize(
+    "invalid_hash",
+    ["tokenizer-v1", None, "ab" * 16 + " " * 32],
+)
 def test_class_udf_state_hash_rejects_invalid_hash(invalid_hash):
     class Configured(Mapper):
         def state_hash(self) -> str:

From ce52ef2ffa2e75136deb3f9a9422a47a023abe3a Mon Sep 17 00:00:00 2001
From: ilongin <ivan.longin1@gmail.com>
Date: Thu, 27 Aug 2026 15:19:19 +0200
Subject: [PATCH 11/16] fixing issue

---
 src/datachain/lib/udf.py   | 22 ++++++++++++----------
 tests/unit/lib/test_udf.py | 22 ++++++++++++++++++++++
 2 files changed, 34 insertions(+), 10 deletions(-)

diff --git a/src/datachain/lib/udf.py b/src/datachain/lib/udf.py
index 291471c4d..bbac80411 100644
--- a/src/datachain/lib/udf.py
+++ b/src/datachain/lib/udf.py
@@ -224,14 +224,17 @@ def prefetch(self) -> int:
         return self.inner.prefetch
 
 
-def _hash_constructor_args(arguments: dict[str, Any]) -> str:
+def _hash_constructor_args(
+    arguments: dict[str, Any], *, warn_on_unsupported: bool = True
+) -> str:
     try:
         return hash_value(arguments)
     except TypeError as exc:
-        logger.warning(
-            "%s; cache reuse across UDF instances is disabled",
-            exc,
-        )
+        if warn_on_unsupported:
+            logger.warning(
+                "%s; cache reuse across UDF instances is disabled",
+                exc,
+            )
         return hashlib.sha256(uuid4().bytes).hexdigest()
 
 
@@ -296,10 +299,6 @@ def process(self, file) -> list[float]:
 
     def __new__(cls, *args, **kwargs):
         instance = super().__new__(cls)
-        if cls.state_hash is not UDFBase.state_hash:
-            # A public state_hash() override owns instance identity, so avoid
-            # normalizing constructor values or warning about opaque arguments.
-            return instance
         try:
             # bound-method signature so `self` is already stripped; correct even
             # when __init__ uses *args instead of a named `self` parameter.
@@ -311,7 +310,10 @@ def __new__(cls, *args, **kwargs):
 
         bound.apply_defaults()
         arguments = cls._constructor_hash_args(dict(bound.arguments))
-        instance._constructor_state_hash = _hash_constructor_args(arguments)
+        instance._constructor_state_hash = _hash_constructor_args(
+            arguments,
+            warn_on_unsupported=cls.state_hash is UDFBase.state_hash,
+        )
         return instance
 
     @classmethod
diff --git a/tests/unit/lib/test_udf.py b/tests/unit/lib/test_udf.py
index 796c8b105..c9509e670 100644
--- a/tests/unit/lib/test_udf.py
+++ b/tests/unit/lib/test_udf.py
@@ -289,6 +289,28 @@ def process(self, x: int) -> int:
     assert "cache reuse across UDF instances is disabled" not in caplog.text
 
 
+def test_class_udf_state_hash_can_extend_automatic_constructor_hash():
+    class Configured(Mapper):
+        def __init__(self, limit: int, extra: str):
+            self.limit = limit
+            self.extra = extra
+
+        def state_hash(self) -> str:
+            return hash_value((super().state_hash(), self.extra))
+
+        def process(self, x: int) -> int:
+            return x + self.limit
+
+    first = Configured(3, "shared")
+    second = Configured(5, "shared")
+    sign_a = get_sign(first, output="y")
+    sign_b = get_sign(second, output="y")
+    udf_a = Mapper._create(sign_a, sign_a.output_schema)
+    udf_b = Mapper._create(sign_b, sign_b.output_schema)
+
+    assert udf_a.hash() != udf_b.hash()
+
+
 @pytest.mark.parametrize(
     "invalid_hash",
     ["tokenizer-v1", None, "ab" * 16 + " " * 32],

From cbcf944ff08d043b5edd34042bb1ea3174d515ee Mon Sep 17 00:00:00 2001
From: ilongin <ivan.longin1@gmail.com>
Date: Thu, 27 Aug 2026 16:43:08 +0200
Subject: [PATCH 12/16] fixing issue

---
 docs/guide/python-engine.md | 16 +++++++++-------
 src/datachain/lib/udf.py    |  3 ++-
 2 files changed, 11 insertions(+), 8 deletions(-)

diff --git a/docs/guide/python-engine.md b/docs/guide/python-engine.md
index bc28cdf90..289f20fc4 100644
--- a/docs/guide/python-engine.md
+++ b/docs/guide/python-engine.md
@@ -159,24 +159,26 @@ receive a unique identity instead, preventing incorrect cache reuse. Override
 `state_hash()` when you can identify such state safely:
 
 ```python
-import hashlib
+from datachain.hash_utils import hash_value
 from datachain.lib.udf import Mapper
 
 class Tokenize(Mapper):
-    def __init__(self, tokenizer, tokenizer_version: str):
+    def __init__(self, tokenizer, tokenizer_id: str, max_length: int):
         self.tokenizer = tokenizer
-        self.tokenizer_version = tokenizer_version
+        self.tokenizer_id = tokenizer_id
+        self.max_length = max_length
 
     def state_hash(self) -> str:
-        return hashlib.sha256(self.tokenizer_version.encode()).hexdigest()
+        return hash_value((self.tokenizer_id, self.max_length))
 
     def process(self, text: str) -> list[str]:
-        return self.tokenizer(text)
+        return self.tokenizer(text)[: self.max_length]
 ```
 
 `state_hash()` must return a SHA-256 hexadecimal string covering all instance state
-that affects output. It replaces automatic constructor hashing; UDF code and schemas
-are still included. An incomplete hash can reuse an incorrect cached result.
+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.
 
diff --git a/src/datachain/lib/udf.py b/src/datachain/lib/udf.py
index bbac80411..b129d6399 100644
--- a/src/datachain/lib/udf.py
+++ b/src/datachain/lib/udf.py
@@ -327,7 +327,8 @@ def state_hash(self) -> str:
         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.
+        default, DataChain returns its automatic constructor-argument hash. Call
+        ``super().state_hash()`` to include that hash in an override.
         """
         return self._constructor_state_hash
 

From dd22402c16b59755f48ed3261c1f16c176fbc019 Mon Sep 17 00:00:00 2001
From: ilongin <ivan.longin1@gmail.com>
Date: Thu, 27 Aug 2026 17:03:11 +0200
Subject: [PATCH 13/16] fixing docs

---
 docs/guide/python-engine.md | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/docs/guide/python-engine.md b/docs/guide/python-engine.md
index 289f20fc4..738bd59df 100644
--- a/docs/guide/python-engine.md
+++ b/docs/guide/python-engine.md
@@ -159,7 +159,8 @@ receive a unique identity instead, preventing incorrect cache reuse. Override
 `state_hash()` when you can identify such state safely:
 
 ```python
-from datachain.hash_utils import hash_value
+import hashlib
+import json
 from datachain.lib.udf import Mapper
 
 class Tokenize(Mapper):
@@ -169,7 +170,11 @@ class Tokenize(Mapper):
         self.max_length = max_length
 
     def state_hash(self) -> str:
-        return hash_value((self.tokenizer_id, self.max_length))
+        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]

From 97e4f430a98281f2ab987d10df57847e088318f0 Mon Sep 17 00:00:00 2001
From: ilongin <ivan.longin1@gmail.com>
Date: Fri, 28 Aug 2026 10:58:07 +0200
Subject: [PATCH 14/16] fixing sorting dicts for hash

---
 src/datachain/hash_utils.py   | 19 ++++++++++---------
 src/datachain/llm/spec.py     |  2 +-
 tests/unit/lib/test_udf.py    | 28 ++++++++++++++++++++++++++++
 tests/unit/test_hash_utils.py |  4 ++--
 4 files changed, 41 insertions(+), 12 deletions(-)

diff --git a/src/datachain/hash_utils.py b/src/datachain/hash_utils.py
index 20a06667f..154aea4ce 100644
--- a/src/datachain/hash_utils.py
+++ b/src/datachain/hash_utils.py
@@ -17,10 +17,13 @@
 ColumnLike: TypeAlias = str | T
 
 
-def normalize_hash_value(value):  # noqa: PLR0911
+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:
@@ -28,24 +31,22 @@ def normalize_hash_value(value):  # noqa: PLR0911
     if value_type in (bool, int, str, bytes):
         return (value_type.__name__, value)
     if value_type is dict:
-        items = (
-            (normalize_hash_value(k), normalize_hash_value(v)) for k, v in value.items()
-        )
-        return ("dict", tuple(sorted(items, key=repr)))
+        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_hash_value, value), key=repr)),
+            tuple(sorted(map(normalize, value), key=repr)),
         )
     if value_type is frozenset:
         return (
             "frozenset",
-            tuple(sorted(map(normalize_hash_value, value), key=repr)),
+            tuple(sorted(map(normalize, value), key=repr)),
         )
     if value_type is list:
-        return ("list", tuple(map(normalize_hash_value, value)))
+        return ("list", tuple(map(normalize, value)))
     if value_type is tuple:
-        return ("tuple", tuple(map(normalize_hash_value, value)))
+        return ("tuple", tuple(map(normalize, value)))
 
     raise TypeError(f"value of type {value_type.__name__!r} cannot be hashed safely")
 
diff --git a/src/datachain/llm/spec.py b/src/datachain/llm/spec.py
index fb2d9fbed..061b9cf13 100644
--- a/src/datachain/llm/spec.py
+++ b/src/datachain/llm/spec.py
@@ -52,7 +52,7 @@ def _without_secrets(value: Any) -> Any:
 
 def _normalize_identity_value(value: Any, fallback: str) -> Any:
     try:
-        return normalize_hash_value(value)
+        return normalize_hash_value(value, sort_dicts=True)
     except TypeError as exc:
         logger.warning("%s; cache reuse for this LLM operation is disabled", exc)
         return ("unsupported", fallback)
diff --git a/tests/unit/lib/test_udf.py b/tests/unit/lib/test_udf.py
index c9509e670..8bb733f81 100644
--- a/tests/unit/lib/test_udf.py
+++ b/tests/unit/lib/test_udf.py
@@ -196,6 +196,34 @@ def process(self, x: int) -> int:
     assert udf_a.hash() == udf_b.hash()
 
 
+@pytest.mark.parametrize(
+    "first_config,second_config",
+    [
+        ({"a": 1, "b": 2}, {"b": 2, "a": 1}),
+        (
+            {"options": {"a": 1, "b": 2}},
+            {"options": {"b": 2, "a": 1}},
+        ),
+    ],
+)
+def test_class_udf_hash_preserves_constructor_dict_order(first_config, second_config):
+    class Configured(Mapper):
+        def __init__(self, config):
+            self.config = config
+
+        def process(self, x: int) -> str:
+            return ",".join(self.config)
+
+    first = Configured(first_config)
+    second = Configured(second_config)
+    sign_a = get_sign(first, output="y")
+    sign_b = get_sign(second, output="y")
+    udf_a = Mapper._create(sign_a, sign_a.output_schema)
+    udf_b = Mapper._create(sign_b, sign_b.output_schema)
+
+    assert udf_a.hash() != udf_b.hash()
+
+
 def test_class_udf_unsupported_constructor_value_disables_cache_reuse(caplog):
     class Opaque:
         def __repr__(self):
diff --git a/tests/unit/test_hash_utils.py b/tests/unit/test_hash_utils.py
index 44839f221..45d5c6189 100644
--- a/tests/unit/test_hash_utils.py
+++ b/tests/unit/test_hash_utils.py
@@ -34,8 +34,8 @@
             (
                 "dict",
                 (
-                    (("str", "x"), ("int", 1)),
                     (("str", "y"), ("int", 2)),
+                    (("str", "x"), ("int", 1)),
                 ),
             ),
         ),
@@ -49,8 +49,8 @@
                         (
                             "dict",
                             (
-                                (("str", "x"), ("int", 1)),
                                 (("str", "y"), ("int", 2)),
+                                (("str", "x"), ("int", 1)),
                             ),
                         ),
                     ),

From 626ac270ddbd0933d05b0808fb1d0f019a716dfe Mon Sep 17 00:00:00 2001
From: ilongin <ivan.longin1@gmail.com>
Date: Fri, 28 Aug 2026 12:05:44 +0200
Subject: [PATCH 15/16] fixing issues

---
 src/datachain/lib/udf.py   |  2 +-
 src/datachain/llm/spec.py  | 13 +++++++++----
 tests/unit/lib/test_llm.py | 10 ++++++++++
 tests/unit/lib/test_udf.py | 21 +++++++++++++++++++++
 4 files changed, 41 insertions(+), 5 deletions(-)

diff --git a/src/datachain/lib/udf.py b/src/datachain/lib/udf.py
index b129d6399..7eb0ecfe4 100644
--- a/src/datachain/lib/udf.py
+++ b/src/datachain/lib/udf.py
@@ -229,7 +229,7 @@ def _hash_constructor_args(
 ) -> str:
     try:
         return hash_value(arguments)
-    except TypeError as exc:
+    except (TypeError, RecursionError) as exc:
         if warn_on_unsupported:
             logger.warning(
                 "%s; cache reuse across UDF instances is disabled",
diff --git a/src/datachain/llm/spec.py b/src/datachain/llm/spec.py
index 061b9cf13..564887420 100644
--- a/src/datachain/llm/spec.py
+++ b/src/datachain/llm/spec.py
@@ -50,10 +50,14 @@ def _without_secrets(value: Any) -> Any:
     return value
 
 
-def _normalize_identity_value(value: Any, fallback: str) -> Any:
+def _normalize_identity_value(
+    value: Any, fallback: str, *, remove_secrets: bool = False
+) -> Any:
     try:
+        if remove_secrets:
+            value = _without_secrets(value)
         return normalize_hash_value(value, sort_dicts=True)
-    except TypeError as exc:
+    except (TypeError, RecursionError) as exc:
         logger.warning("%s; cache reuse for this LLM operation is disabled", exc)
         return ("unsupported", fallback)
 
@@ -187,7 +191,6 @@ def identity(self, model: str, llm_params: Any = None) -> tuple:
         params = self.params
         if isinstance(llm_params, dict):
             params = {**llm_params, **self.params}
-        params = _without_secrets(params)
         return (
             self.kind,
             model,
@@ -198,7 +201,9 @@ def identity(self, model: str, llm_params: Any = None) -> tuple:
             self.context_col,
             self.type,
             self.include_usage,
-            _normalize_identity_value(params, self._identity_fallback),
+            _normalize_identity_value(
+                params, self._identity_fallback, remove_secrets=True
+            ),
         )
 
     def _resolve_model(self, settings: "Settings") -> str:
diff --git a/tests/unit/lib/test_llm.py b/tests/unit/lib/test_llm.py
index ca2b78419..7586e6e21 100644
--- a/tests/unit/lib/test_llm.py
+++ b/tests/unit/lib/test_llm.py
@@ -745,6 +745,16 @@ class Opaque:
     assert spec.identity("m") == spec.identity("m")
 
 
+def test_cyclic_param_value_disables_cache_reuse():
+    options = {}
+    options["self"] = options
+    first = llm.complete("t", options=options)
+    second = llm.complete("t", options=options)
+
+    assert first.identity("m") == first.identity("m")
+    assert first.identity("m") != second.identity("m")
+
+
 def test_stable_param_values_have_stable_identity():
     first = llm.complete("t", temperature=0.0, opt={"a": 1}).identity("m")
     second = llm.complete("t", temperature=0.0, opt={"a": 1}).identity("m")
diff --git a/tests/unit/lib/test_udf.py b/tests/unit/lib/test_udf.py
index 8bb733f81..04526be9c 100644
--- a/tests/unit/lib/test_udf.py
+++ b/tests/unit/lib/test_udf.py
@@ -282,6 +282,27 @@ def process(self, x: int) -> int:
     assert udf_a.hash() != udf_b.hash()
 
 
+def test_class_udf_cyclic_constructor_value_disables_cache_reuse():
+    class Configured(Mapper):
+        def __init__(self, config):
+            self.config = config
+
+        def process(self, x: int) -> int:
+            return x
+
+    config = {}
+    config["self"] = config
+    first = Configured(config)
+    second = Configured(config)
+    sign_a = get_sign(first, output="y")
+    sign_b = get_sign(second, output="y")
+    udf_a = Mapper._create(sign_a, sign_a.output_schema)
+    udf_b = Mapper._create(sign_b, sign_b.output_schema)
+
+    assert udf_a.hash() == udf_a.hash()
+    assert udf_a.hash() != udf_b.hash()
+
+
 @pytest.mark.parametrize(
     "first_key,second_key,matches",
     [

From 615bc8a5d4b1c0488700fc02b46c7466f26d15e3 Mon Sep 17 00:00:00 2001
From: ilongin <ivan.longin1@gmail.com>
Date: Fri, 28 Aug 2026 12:25:50 +0200
Subject: [PATCH 16/16] fixing output_schema hash

---
 src/datachain/lib/arrow.py        | 10 +++++-----
 src/datachain/lib/dc/datachain.py |  3 +++
 tests/unit/lib/test_arrow.py      |  1 +
 tests/unit/test_datachain_hash.py | 27 ++++++++++++++++++++++++++-
 4 files changed, 35 insertions(+), 6 deletions(-)

diff --git a/src/datachain/lib/arrow.py b/src/datachain/lib/arrow.py
index 2aee52cec..0d9dac38f 100644
--- a/src/datachain/lib/arrow.py
+++ b/src/datachain/lib/arrow.py
@@ -76,6 +76,7 @@ def __init__(
         output_schema: type["BaseModel"] | None = None,
         source: bool = True,
         nrows: int | None = None,
+        _generated_output_schema: bool = False,
         **kwargs,
     ):
         """
@@ -99,12 +100,11 @@ def __init__(
 
     @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)
+        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()
diff --git a/src/datachain/lib/dc/datachain.py b/src/datachain/lib/dc/datachain.py
index 9a259924a..0aa08cddf 100644
--- a/src/datachain/lib/dc/datachain.py
+++ b/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,
             ),
diff --git a/tests/unit/lib/test_arrow.py b/tests/unit/lib/test_arrow.py
index 582ed9238..2fb57b162 100644
--- a/tests/unit/lib/test_arrow.py
+++ b/tests/unit/lib/test_arrow.py
@@ -32,6 +32,7 @@ def make_generator(output_schema, nrows=None):
             input_schema=input_schema,
             output_schema=output_schema,
             nrows=nrows,
+            _generated_output_schema=True,
             parse_options=parse_options,
             format=CsvFileFormat(parse_options=parse_options),
         )
diff --git a/tests/unit/test_datachain_hash.py b/tests/unit/test_datachain_hash.py
index e9ac86bee..55a3691f0 100644
--- a/tests/unit/test_datachain_hash.py
+++ b/tests/unit/test_datachain_hash.py
@@ -1,6 +1,6 @@
 import pandas as pd
 import pytest
-from pydantic import BaseModel
+from pydantic import BaseModel, field_validator
 
 import datachain as dc
 from datachain import func
@@ -107,6 +107,31 @@ def test_read_csv_single_file_is_deterministic(test_session, tmp_dir):
     assert h1 == h2
 
 
+def test_read_csv_user_output_model_behavior_changes_hash(test_session, tmp_dir):
+    def make_model(uppercase):
+        class Row(BaseModel):
+            value: str
+
+            @field_validator("value")
+            @classmethod
+            def transform(cls, value):
+                return value.upper() if uppercase else value.lower()
+
+        return Row
+
+    path = tmp_dir / "test.csv"
+    pd.DataFrame({"value": ["MiXeD"]}).to_csv(path, index=False)
+
+    uppercase = dc.read_csv(
+        path.as_uri(), output=make_model(True), session=test_session
+    )
+    lowercase = dc.read_csv(
+        path.as_uri(), output=make_model(False), session=test_session
+    )
+
+    assert uppercase._query.hash() != lowercase._query.hash()
+
+
 def test_read_csv_multi_file_glob_is_deterministic(test_session, tmp_dir):
     pd.DataFrame({"a": [1, 2]}).to_csv(tmp_dir / "a.csv", index=False)
     pd.DataFrame({"a": [3, 4]}).to_csv(tmp_dir / "b.csv", index=False)