fix: store all DB hash columns as large_binary (ITL-539)

DESIGN_ISSUES.md

@@ -1264,3 +1264,22 @@ across this version range.
 
 **Ongoing:** pyspiral releases frequently. See PLT-1785 for the tracking issue
 covering routine version bumps.
+
+---
+
+## `src/orcapod/core/nodes/operator_node.py`
+
+### ON1 — OperatorJobNode has no v0→v1 schema migration for `NODE_CONTENT_HASH_COL`
+**Status:** open
+**Severity:** high
+**Issue:** ITL-539
+
+`OperatorJobNode` stores `NODE_CONTENT_HASH_COL` as `large_binary` (changed in ITL-539),
+but unlike `FunctionJobNode` and `ResultCache`, it has no `_ensure_schema()` guard and no
+v0→v1 migration utility. Any existing pipeline DB written by the old code that stored
+`NODE_CONTENT_HASH_COL` as `large_string` will fail with an Arrow schema mismatch on the
+next write attempt. This is accepted as a breaking change for a pre-v0.1.0 project; users
+must drop and recreate the affected pipeline DB tables manually.
+
+A proper migration utility (analogous to `migrate_pipeline_v0_to_v1`) should be implemented
+before v0.1.0 ship.

src/orcapod/core/data_function.py

@@ -463,10 +463,10 @@ def __init__(
         semantic_hasher = self.data_context.semantic_hasher
         self._function_signature_hash = semantic_hasher.hash_object(
             get_function_signature(function)
-        ).to_string()
+        ).to_prefixed_digest()
         self._function_content_hash = semantic_hasher.hash_object(
             get_function_components(self._function)
-        ).to_string()
+        ).to_prefixed_digest()
 
     @property
     def canonical_function_name(self) -> str:
@@ -486,8 +486,8 @@ def get_function_variation_data_schema(self) -> Schema:
         """Schema for the data returned by ``get_function_variation_data``."""
         return Schema({
             "function_name": str,
-            "function_signature_hash": str,
-            "function_content_hash": str,
+            "function_signature_hash": bytes,
+            "function_content_hash": bytes,
             "git_hash": str,
         })
 

src/orcapod/core/nodes/operator_node.py

@@ -775,7 +775,16 @@ def _filter_by_content_hash(self, table: pa.Table) -> pa.Table:
                 f"required column {col_name!r} is missing from the stored table. "
                 "This may indicate records written by an older version of the code."
             )
-        own_hash = self.content_hash().to_string()
+        col_type = table.schema.field(col_name).type
+        if col_type != pa.large_binary():
+            raise ValueError(
+                f"Cannot isolate records for table_scope='pipeline_hash': "
+                f"column {col_name!r} has type {col_type!r}, expected large_binary. "
+                "This table was written by an older version of the code that stored "
+                "this column as large_string. Drop and recreate the affected pipeline "
+                "DB tables (see DESIGN_ISSUES.md ON1)."
+            )
+        own_hash = self.content_hash().to_prefixed_digest()
         mask = pc.equal(table.column(col_name), own_hash)
         return table.filter(mask)
 
@@ -801,7 +810,10 @@ def _store_output_stream(self, stream: StreamProtocol) -> None:
         n_rows = output_table.num_rows
         output_table = output_table.append_column(
             constants.NODE_CONTENT_HASH_COL,
-            pa.repeat(self.content_hash().to_string(), n_rows).cast(pa.large_string()),
+            pa.array(
+                [self.content_hash().to_prefixed_digest()] * n_rows,
+                type=pa.large_binary(),
+            ),
         )
 
         # Per-row record hashes for dedup: hash(tag + data + system_tags + node_content_hash).

src/orcapod/core/result_cache.py

@@ -10,13 +10,12 @@
 import logging
 import uuid
 from datetime import datetime, timezone
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from orcapod.errors import SchemaVersionError
 from orcapod.protocols.core_protocols import DataProtocol
 from orcapod.protocols.database_protocols import ArrowDatabaseProtocol
 from orcapod.system_constants import constants, RESULT_DB_SCHEMA_VERSION
-from orcapod.types import ContentHash
 from orcapod.utils.lazy_module import LazyModule
 
 if TYPE_CHECKING:
@@ -29,20 +28,6 @@
 logger = logging.getLogger(__name__)
 
 
-def _hash_val_to_binary(val: "str | bytes | memoryview | None") -> "bytes | None":
-    """Convert a ContentHash value to its prefixed binary digest.
-
-    Tolerates both ``str`` (v0 format, passed through ``ContentHash.from_string``)
-    and ``bytes``/``memoryview`` (already binary v1 format, returned as-is).
-    Returns ``None`` for ``None`` inputs.
-    """
-    if val is None:
-        return None
-    if isinstance(val, (bytes, memoryview)):
-        return bytes(val)
-    return ContentHash.from_string(val).to_prefixed_digest()
-
-
 # Process-level cache of v1 result DB paths that have already been checked for
 # legacy v0 schema. Populated on first access; prevents repeated table_exists
 # calls for the same path within a single process.
@@ -169,7 +154,7 @@ def _ensure_rdb_schema(self) -> None:
     def lookup(
         self,
         input_data: DataProtocol,
-        additional_constraints: dict[str, str] | None = None,
+        additional_constraints: dict[str, Any] | None = None,
     ) -> DataProtocol | None:
         """Look up a cached output data for *input_data*.
 
@@ -183,7 +168,9 @@ def lookup(
             input_data: The input data whose content hash is the
                 primary lookup key.
             additional_constraints: Optional extra column-value pairs to
-                include in the lookup query.
+                include in the lookup query. Values may be ``bytes`` (for
+                binary hash columns) or other scalar types (e.g. ``str``
+                for ``function_name``).
 
         Returns:
             The cached output data with ``RESULT_COMPUTED_FLAG: False``
@@ -195,7 +182,7 @@ def lookup(
 
         RECORD_ID_COL = "_record_id"
 
-        constraints: dict[str, bytes] = {
+        constraints: dict[str, Any] = {
             constants.INPUT_DATA_HASH_COL: input_data.content_hash().to_prefixed_digest(),
         }
         if additional_constraints:
@@ -282,22 +269,6 @@ def store(
             )
             col_idx += 1
 
-        # Convert ContentHash variation columns to large_binary (v1 schema).
-        # Tolerates both str (v0 format) and bytes/memoryview (already binary — pass through).
-        _HASH_VAR_COLS = {
-            f"{constants.PF_VARIATION_PREFIX}function_signature_hash",
-            f"{constants.PF_VARIATION_PREFIX}function_content_hash",
-        }
-        for col_name in _HASH_VAR_COLS:
-            if col_name in data_table.column_names:
-                col_idx = data_table.column_names.index(col_name)
-                raw_vals = data_table.column(col_name).to_pylist()
-                binary_vals = pa.array(
-                    [_hash_val_to_binary(v) for v in raw_vals],
-                    type=pa.large_binary(),
-                )
-                data_table = data_table.set_column(col_idx, col_name, binary_vals)
-
         # Add input data hash as large_binary at position 0 (v1 schema).
         data_table = data_table.add_column(
             0,

src/orcapod/side_effects.py

@@ -267,7 +267,7 @@ def _execute_side_effect_row(
                 pipeline_database=pipeline_database,
                 table_path=table_path,
                 record_id=record_id,
-                record_id_hash_str=record_id_hash.to_string(),
+                record_id_hash_bytes=record_id_hash.to_prefixed_digest(),
                 run_id=run_id,
             )
         return (tag, data)
@@ -287,7 +287,7 @@ def _write_invocation_row(
     pipeline_database: ArrowDatabaseProtocol,
     table_path: tuple[str, ...],
     record_id: bytes,
-    record_id_hash_str: str,
+    record_id_hash_bytes: bytes,
     run_id: str | None,
 ) -> None:
     """Write one success row to the side-effect invocation log table.
@@ -301,15 +301,15 @@ def _write_invocation_row(
         table_path: Path tuple for the invocation log table.
         record_id: Deterministic bytes key for this ``(input, pod version)``
             pair — the prefixed digest of the unified preimage hash.
-        record_id_hash_str: String form of the record-ID hash (stored for
-            human inspection).
+        record_id_hash_bytes: Binary prefixed digest of the record-ID hash
+            (stored for human inspection via ``ContentHash.from_prefixed_digest``).
         run_id: Pipeline run identifier (or ``None`` for standalone mode).
     """
     executed_at = datetime.datetime.now(datetime.timezone.utc)
     record = pa.table(
         {
             "record_id_hash": pa.array(
-                [record_id_hash_str], type=pa.large_string()
+                [record_id_hash_bytes], type=pa.large_binary()
             ),
             "pipeline_run_id": pa.array(
                 [run_id], type=pa.large_string()

superpowers/specs/2026-07-21-hash-columns-large-binary-design.md

@@ -0,0 +1,189 @@
+# Hash Columns as `large_binary` — Final Cleanup (ITL-539)
+
+**Date:** 2026-07-21
+**Issue:** [ITL-539](https://linear.app/metamorphic/issue/ITL-539)
+
+## Overview
+
+Most of ITL-539 was already implemented in earlier PRs (ITL-534, ITL-535): the pipeline DB and
+result cache now write `INPUT_DATA_HASH_COL`, `OUTPUT_DATA_HASH_COL`, and `NODE_CONTENT_HASH_COL`
+as `large_binary`, and full v0→v1 migration utilities exist for both databases.
+
+This spec covers the remaining four gaps identified by a full audit:
+
+1. `data_function.py` — variation hash fields still produced as `large_string`
+2. `result_cache.py` — backward-compat conversion shim, no longer needed
+3. `operator_node.py` — `NODE_CONTENT_HASH_COL` stored as `large_string`
+4. `side_effects.py` — `record_id_hash` log column stored as `large_string`
+
+## Audit Summary
+
+### Already correct (no changes needed)
+
+| Component | Column | Type |
+|-----------|--------|------|
+| `FunctionJobNode.add_pipeline_record()` | `INPUT_DATA_HASH_COL`, `OUTPUT_DATA_HASH_COL`, `NODE_CONTENT_HASH_COL` | `large_binary` ✅ |
+| `ResultCache.store()` | `INPUT_DATA_HASH_COL` | `large_binary` ✅ |
+| `ResultCache.lookup()` | query uses `to_prefixed_digest()` | ✅ |
+| `ContentHash.from_prefixed_digest()` | exists in `types.py` | ✅ |
+| v0→v1 migrations (pipeline DB + result DB) | all hash columns | ✅ |
+
+### Non-DB string usages (intentional, out of scope)
+
+- Path components: `f"schema:{hash.to_string()}"`, `f"instance:{hash.to_string()}"` — string path keys, not Arrow columns
+- Observer calls: `on_node_start(label, hash_str)` — observer interface uses strings
+- Python dict keys: `content_hash().to_string()` used as in-memory graph keys
+- Source IDs: `f"...:{hash.to_string()}"` — string identifier, not stored in DB
+- User-facing `_content_hash` column in `as_table()` — optional display column, not stored in DB
+- Semantic hasher internals — hashing preimage computation
+
+## Changes
+
+### 1. `src/orcapod/core/data_function.py`
+
+**Problem:** `_function_signature_hash` and `_function_content_hash` are computed and stored as
+`str` via `to_string()`. The schema declares them as `str`, so `Datagram.as_table()` emits them
+as `large_string` columns.
+
+**Fix:**
+- Change both `to_string()` calls to `to_prefixed_digest()` — the values become `bytes`.
+- Change both schema entries from `str` to `bytes` — `Datagram.as_table()` will emit
+  `large_binary` columns via the `bytes → pa.large_binary()` mapping in `universal_converter.py`.
+
+```python
+# Before
+self._function_signature_hash = semantic_hasher.hash_object(
+    get_function_signature(function)
+).to_string()
+self._function_content_hash = semantic_hasher.hash_object(
+    get_function_components(self._function)
+).to_string()
+
+# After
+self._function_signature_hash = semantic_hasher.hash_object(
+    get_function_signature(function)
+).to_prefixed_digest()
+self._function_content_hash = semantic_hasher.hash_object(
+    get_function_components(self._function)
+).to_prefixed_digest()
+```
+
+```python
+# Schema before
+def get_function_variation_data_schema(self) -> Schema:
+    return Schema({
+        "function_name": str,
+        "function_signature_hash": str,   # ❌
+        "function_content_hash": str,     # ❌
+        "git_hash": str,
+    })
+
+# Schema after
+def get_function_variation_data_schema(self) -> Schema:
+    return Schema({
+        "function_name": str,
+        "function_signature_hash": bytes,  # ✅
+        "function_content_hash": bytes,    # ✅
+        "git_hash": str,
+    })
+```
+
+**No other callers** access `_function_signature_hash` or `_function_content_hash` directly —
+both are only read through `get_function_variation_data()`.
+
+### 2. `src/orcapod/core/result_cache.py`
+
+**Problem:** The `_hash_val_to_binary()` helper and the `_HASH_VAR_COLS` conversion loop in
+`store()` were added as a tolerance shim to handle string-typed variation hash columns. With
+`PythonDataFunction` now producing bytes, the shim is dead code.
+
+**Fix:** Remove `_hash_val_to_binary()` and the entire `_HASH_VAR_COLS` conversion loop from
+`store()`. The variation datagram's `as_table()` now emits `large_binary` columns directly.
+
+Note: The `constraints` dict in `lookup()` uses `bytes` values (via `to_prefixed_digest()`). Its
+type annotation `dict[str, str]` is incorrect and should be corrected to `dict[str, bytes]`.
+
+### 3. `src/orcapod/core/nodes/operator_node.py`
+
+**Problem:** `_store_output_stream()` appends `NODE_CONTENT_HASH_COL` as `large_string` via
+`.to_string()`. `_filter_by_content_hash()` reads it back with `.to_string()` and a string
+`pc.equal` comparison.
+
+**Fix:**
+- Write: change `to_string()` / `large_string()` to `to_prefixed_digest()` / `large_binary()`.
+- Read: change filter to compare against the binary digest.
+
+```python
+# Write — before
+pa.repeat(self.content_hash().to_string(), n_rows).cast(pa.large_string())
+
+# Write — after
+pa.array(
+    [self.content_hash().to_prefixed_digest()] * n_rows,
+    type=pa.large_binary(),
+)
+```
+
+```python
+# Filter — before
+own_hash = self.content_hash().to_string()
+mask = pc.equal(table.column(col_name), own_hash)
+
+# Filter — after
+own_hash = self.content_hash().to_prefixed_digest()
+mask = pc.equal(table.column(col_name), own_hash)
+```
+
+**Breaking change note:** `OperatorJobNode` has no schema versioning infrastructure. Any existing
+operator node pipeline DB with the old `large_string` schema will fail on next write due to schema
+mismatch. This is acceptable for a pre-v0.1.0 project. A DESIGN_ISSUES.md note will document the
+migration gap.
+
+### 4. `src/orcapod/side_effects.py`
+
+**Problem:** `_write_invocation_row()` stores the `record_id_hash` column as `large_string` using
+`.to_string()`. The binary `record_id` is already the primary key; `record_id_hash` is a display
+column stored alongside it.
+
+**Fix:**
+- Change the function signature: `record_id_hash_str: str` → `record_id_hash_bytes: bytes`.
+- Change the call site: pass `record_id_hash.to_prefixed_digest()` instead of `.to_string()`.
+- Change the Arrow array: `pa.large_string()` → `pa.large_binary()`.
+
+No migration needed — `record_id_hash` is not used for record lookup; each invocation writes a new
+row and the old column value is never re-read by column comparison.
+
+## Tests
+
+### New tests needed
+
+**`tests/test_pipeline/test_result_cache.py`** (or extend existing):
+- Verify `ResultCache.store()` stores `function_signature_hash` and `function_content_hash` as
+  `large_binary` columns when used with a `PythonDataFunction`.
+- Verify the stored bytes decode correctly via `ContentHash.from_prefixed_digest()`.
+
+**`tests/test_pipeline/test_operator_node_hash.py`** (or extend existing operator node tests):
+- Verify `OperatorJobNode._store_output_stream()` writes `NODE_CONTENT_HASH_COL` as `large_binary`.
+- Verify `_filter_by_content_hash()` correctly isolates rows by binary hash value.
+
+**`tests/test_pipeline/test_side_effects.py`** (or extend existing):
+- Verify `_write_invocation_row()` stores `record_id_hash` as `large_binary`.
+- Verify the stored bytes decode correctly via `ContentHash.from_prefixed_digest()`.
+
+### Existing tests
+
+- All 4011 existing tests must continue to pass after the changes.
+- `tests/test_migrations/` — no changes; existing migration tests cover string-to-binary
+  conversion for the result DB and pipeline DB.
+
+## Out of Scope
+
+Per the original issue and confirmed in design:
+
+- `_PIPELINE_BASE_ENTRY_ID_COL` — already `large_binary`
+- `DATA_RECORD_ID` — already `large_binary`
+- `NODE_CONTENT_HASH_COL` and `DATA_RECORD_ID` — out of scope per original issue (except
+  the `OperatorJobNode` fix covered above, which was found during the audit)
+- Backfilling old tag rows — tracked by ITL-535
+- Non-DB string uses of `to_string()` (path components, observer calls, dict keys)
+- User-facing `_content_hash` display column in `FunctionJobNode.as_table()`