From 63fad8f69ee0cf484bfee9a7c44cf36eff5ddbed Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 21 Jul 2026 21:45:54 +0000 Subject: [PATCH 1/9] docs(spec): add ITL-539 hash columns large_binary final cleanup design Co-Authored-By: Claude Sonnet 4.6 --- ...-07-21-hash-columns-large-binary-design.md | 189 ++++++++++++++++++ 1 file changed, 189 insertions(+) create mode 100644 superpowers/specs/2026-07-21-hash-columns-large-binary-design.md diff --git a/superpowers/specs/2026-07-21-hash-columns-large-binary-design.md b/superpowers/specs/2026-07-21-hash-columns-large-binary-design.md new file mode 100644 index 00000000..9bfb2777 --- /dev/null +++ b/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()` From e2b23a945e965ef169e544bdba188a96c22d5ad0 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 21 Jul 2026 22:27:00 +0000 Subject: [PATCH 2/9] fix(data_function): store variation hashes as large_binary bytes (ITL-539) Switch _function_signature_hash and _function_content_hash from to_string() (str) to to_prefixed_digest() (bytes) in PythonDataFunction, and update get_function_variation_data_schema() to declare those fields as bytes. Add TestVariationHashSchema covering the new behaviour and update the now-stale test_all_values_are_strings assertion. --- src/orcapod/core/data_function.py | 8 ++--- .../data_function/test_data_function.py | 31 +++++++++++++++++-- 2 files changed, 32 insertions(+), 7 deletions(-) diff --git a/src/orcapod/core/data_function.py b/src/orcapod/core/data_function.py index 0d136517..f6c05d0a 100644 --- a/src/orcapod/core/data_function.py +++ b/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, }) diff --git a/tests/test_core/data_function/test_data_function.py b/tests/test_core/data_function/test_data_function.py index 15441fb6..cd204d42 100644 --- a/tests/test_core/data_function/test_data_function.py +++ b/tests/test_core/data_function/test_data_function.py @@ -365,10 +365,10 @@ def test_returns_expected_keys(self, add_pf): "git_hash", } - def test_all_values_are_strings(self, add_pf): + def test_non_hash_values_are_strings(self, add_pf): data = add_pf.get_function_variation_data() - for k, v in data.items(): - assert isinstance(v, str), f"Value for '{k}' is not a string: {v!r}" + assert isinstance(data["function_name"], str) + assert isinstance(data["git_hash"], str) def test_function_name_matches_canonical(self, add_pf): data = add_pf.get_function_variation_data() @@ -814,3 +814,28 @@ def foo(x: str) -> str: h2 = self._sig_hash(foo) assert h1 != h2 + + +class TestVariationHashSchema: + def test_function_signature_hash_is_bytes(self, add_pf): + """PythonDataFunction stores variation hashes as bytes (-> large_binary).""" + variation = add_pf.get_function_variation_data() + assert isinstance(variation["function_signature_hash"], bytes) + + def test_function_content_hash_is_bytes(self, add_pf): + variation = add_pf.get_function_variation_data() + assert isinstance(variation["function_content_hash"], bytes) + + def test_variation_schema_has_bytes_types(self, add_pf): + schema = add_pf.get_function_variation_data_schema() + assert schema["function_signature_hash"] is bytes + assert schema["function_content_hash"] is bytes + + def test_variation_hash_decodes_to_content_hash(self, add_pf): + from orcapod.types import ContentHash + variation = add_pf.get_function_variation_data() + sig_hash = ContentHash.from_prefixed_digest( + variation["function_signature_hash"] + ) + assert sig_hash.method == "semantic_v0.1" + assert len(sig_hash.digest) == 32 From 471b7b2b85ff6d3128b6d873a7515a51c99028f9 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 00:13:46 +0000 Subject: [PATCH 3/9] test(data_function): fix minor quality issues in TestVariationHashSchema (ITL-539) --- tests/test_core/data_function/test_data_function.py | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/tests/test_core/data_function/test_data_function.py b/tests/test_core/data_function/test_data_function.py index cd204d42..ba1d9332 100644 --- a/tests/test_core/data_function/test_data_function.py +++ b/tests/test_core/data_function/test_data_function.py @@ -19,6 +19,7 @@ from orcapod.core.datagrams import Data from orcapod.core.data_function import PythonDataFunction, parse_function_outputs from orcapod.protocols.core_protocols import DataFunctionProtocol +from orcapod.types import ContentHash # --------------------------------------------------------------------------- # Helpers @@ -823,6 +824,7 @@ def test_function_signature_hash_is_bytes(self, add_pf): assert isinstance(variation["function_signature_hash"], bytes) def test_function_content_hash_is_bytes(self, add_pf): + """PythonDataFunction stores content hashes as bytes (-> large_binary).""" variation = add_pf.get_function_variation_data() assert isinstance(variation["function_content_hash"], bytes) @@ -832,10 +834,9 @@ def test_variation_schema_has_bytes_types(self, add_pf): assert schema["function_content_hash"] is bytes def test_variation_hash_decodes_to_content_hash(self, add_pf): - from orcapod.types import ContentHash + """Both variation hash bytes round-trip through ContentHash.from_prefixed_digest.""" variation = add_pf.get_function_variation_data() - sig_hash = ContentHash.from_prefixed_digest( - variation["function_signature_hash"] - ) - assert sig_hash.method == "semantic_v0.1" - assert len(sig_hash.digest) == 32 + sig_hash = ContentHash.from_prefixed_digest(variation["function_signature_hash"]) + content_hash = ContentHash.from_prefixed_digest(variation["function_content_hash"]) + assert isinstance(sig_hash, ContentHash) + assert isinstance(content_hash, ContentHash) From 36bd0fbc31d09ede55b180367870641ea9cc298d Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 00:17:20 +0000 Subject: [PATCH 4/9] fix(result_cache): remove string-hash tolerance shim, fix lookup annotation (ITL-539) Co-Authored-By: Claude Sonnet 4.6 --- src/orcapod/core/result_cache.py | 33 +--------------------- tests/test_core/test_result_cache.py | 42 ++++++++++++++++++++++++++++ 2 files changed, 43 insertions(+), 32 deletions(-) diff --git a/src/orcapod/core/result_cache.py b/src/orcapod/core/result_cache.py index 0b1df078..91fca68f 100644 --- a/src/orcapod/core/result_cache.py +++ b/src/orcapod/core/result_cache.py @@ -16,7 +16,6 @@ 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, bytes] | None = None, ) -> DataProtocol | None: """Look up a cached output data for *input_data*. @@ -282,22 +267,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, diff --git a/tests/test_core/test_result_cache.py b/tests/test_core/test_result_cache.py index 813a0ecd..794002c6 100644 --- a/tests/test_core/test_result_cache.py +++ b/tests/test_core/test_result_cache.py @@ -397,3 +397,45 @@ def test_empty_data_cache_miss(self): result = cache.lookup(empty_data) assert result is None + + +# --------------------------------------------------------------------------- +# Hash column types +# --------------------------------------------------------------------------- + + +class TestStoreHashColumnTypes: + def test_variation_signature_hash_is_large_binary(self): + cache, db = _make_cache() + pf = _make_pf() + _compute_and_store(cache, pf, Data({"x": 10})) + + records = db.get_all_records(cache.record_path) + assert records is not None + sig_col = f"{constants.PF_VARIATION_PREFIX}function_signature_hash" + assert sig_col in records.column_names + assert records.schema.field(sig_col).type == pa.large_binary() + + def test_variation_content_hash_is_large_binary(self): + cache, db = _make_cache() + pf = _make_pf() + _compute_and_store(cache, pf, Data({"x": 10})) + + records = db.get_all_records(cache.record_path) + assert records is not None + content_col = f"{constants.PF_VARIATION_PREFIX}function_content_hash" + assert content_col in records.column_names + assert records.schema.field(content_col).type == pa.large_binary() + + def test_variation_hash_bytes_decode_to_content_hash(self): + cache, db = _make_cache() + pf = _make_pf() + _compute_and_store(cache, pf, Data({"x": 10})) + + records = db.get_all_records(cache.record_path) + assert records is not None + sig_col = f"{constants.PF_VARIATION_PREFIX}function_signature_hash" + raw_bytes = records.column(sig_col)[0].as_py() + assert isinstance(raw_bytes, bytes) + ch = ContentHash.from_prefixed_digest(raw_bytes) + assert isinstance(ch, ContentHash) From 1db4275aab806b754a84b448f7cae40658d4458a Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 00:28:35 +0000 Subject: [PATCH 5/9] fix(operator_node): store NODE_CONTENT_HASH_COL as large_binary (ITL-539) Co-Authored-By: Claude Sonnet 4.6 --- src/orcapod/core/nodes/operator_node.py | 7 ++- .../test_core/operators/test_operator_node.py | 43 +++++++++++++++++++ 2 files changed, 48 insertions(+), 2 deletions(-) diff --git a/src/orcapod/core/nodes/operator_node.py b/src/orcapod/core/nodes/operator_node.py index 84984877..f5cf09ba 100644 --- a/src/orcapod/core/nodes/operator_node.py +++ b/src/orcapod/core/nodes/operator_node.py @@ -775,7 +775,7 @@ 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() + own_hash = self.content_hash().to_prefixed_digest() mask = pc.equal(table.column(col_name), own_hash) return table.filter(mask) @@ -801,7 +801,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). diff --git a/tests/test_core/operators/test_operator_node.py b/tests/test_core/operators/test_operator_node.py index e3783075..bd75e040 100644 --- a/tests/test_core/operators/test_operator_node.py +++ b/tests/test_core/operators/test_operator_node.py @@ -443,3 +443,46 @@ def test_repr(self, simple_stream): node = _make_node(op, (simple_stream,)) r = repr(node) assert "OperatorJobNode" in r + + +# --------------------------------------------------------------------------- +# Hash column type +# --------------------------------------------------------------------------- + + +class TestOperatorNodeHashColumnType: + def test_node_content_hash_col_is_large_binary(self, simple_stream): + """NODE_CONTENT_HASH_COL must be stored as large_binary in the pipeline DB.""" + import pyarrow as pa + from orcapod.system_constants import constants + + db = InMemoryArrowDatabase() + op = MapData({"x": "renamed_x"}) + node = OperatorJobNode( + operator=op, + input_streams=(simple_stream,), + pipeline_database=db, + cache_mode=CacheMode.LOG, + ) + node.run() + + all_records = db.get_all_records(node.node_identity_path) + assert all_records is not None + col_name = constants.NODE_CONTENT_HASH_COL + assert col_name in all_records.column_names + assert all_records.schema.field(col_name).type == pa.large_binary() + + def test_filter_by_content_hash_works_with_binary(self, simple_stream): + """After storing, iter_data() returns only rows for this node's hash.""" + db = InMemoryArrowDatabase() + op = MapData({"x": "renamed_x"}) + node = OperatorJobNode( + operator=op, + input_streams=(simple_stream,), + pipeline_database=db, + cache_mode=CacheMode.LOG, + ) + node.run() + + rows = list(node.iter_data()) + assert len(rows) == 3 # simple_stream has 3 rows From 69a1565288928939528fbc050d6cd168e5976831 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 00:37:26 +0000 Subject: [PATCH 6/9] test(operator_node): move inline imports to module level (ITL-539) --- tests/test_core/operators/test_operator_node.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tests/test_core/operators/test_operator_node.py b/tests/test_core/operators/test_operator_node.py index bd75e040..d9fb7913 100644 --- a/tests/test_core/operators/test_operator_node.py +++ b/tests/test_core/operators/test_operator_node.py @@ -27,6 +27,7 @@ from orcapod.databases import InMemoryArrowDatabase from orcapod.protocols.core_protocols import StreamProtocol from orcapod.protocols.hashing_protocols import PipelineElementProtocol +from orcapod.system_constants import constants from orcapod.types import CacheMode @@ -453,9 +454,6 @@ def test_repr(self, simple_stream): class TestOperatorNodeHashColumnType: def test_node_content_hash_col_is_large_binary(self, simple_stream): """NODE_CONTENT_HASH_COL must be stored as large_binary in the pipeline DB.""" - import pyarrow as pa - from orcapod.system_constants import constants - db = InMemoryArrowDatabase() op = MapData({"x": "renamed_x"}) node = OperatorJobNode( From 7274a13b61afae0a4a8b967242f69f2b5e27831e Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 00:55:13 +0000 Subject: [PATCH 7/9] fix(side_effects): store record_id_hash as large_binary (ITL-539) Co-Authored-By: Claude Sonnet 4.6 --- src/orcapod/side_effects.py | 10 ++-- .../side_effect_pod/test_side_effect_pod.py | 52 +++++++++++++++++++ 2 files changed, 57 insertions(+), 5 deletions(-) diff --git a/src/orcapod/side_effects.py b/src/orcapod/side_effects.py index 8b272cda..0958837c 100644 --- a/src/orcapod/side_effects.py +++ b/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() diff --git a/tests/test_core/side_effect_pod/test_side_effect_pod.py b/tests/test_core/side_effect_pod/test_side_effect_pod.py index d29653ea..5dc47805 100644 --- a/tests/test_core/side_effect_pod/test_side_effect_pod.py +++ b/tests/test_core/side_effect_pod/test_side_effect_pod.py @@ -742,3 +742,55 @@ def fn(value, ctx): stream = _make_stream(1) with pytest.raises(ValueError, match="collides"): list(pod.process(stream).iter_data()) + + +class TestSideEffectInvocationLogHashType: + def test_record_id_hash_is_large_binary(self): + """_write_invocation_row stores record_id_hash as large_binary.""" + import pyarrow as pa + from orcapod.databases import InMemoryArrowDatabase + from orcapod.side_effects import _write_invocation_row + + db = InMemoryArrowDatabase() + table_path = ("test", "invocation_log") + record_id = b"\xab" * 16 + record_id_hash_bytes = b"\x00sha256:" + bytes(range(32)) # arbitrary bytes + + _write_invocation_row( + pipeline_database=db, + table_path=table_path, + record_id=record_id, + record_id_hash_bytes=record_id_hash_bytes, + run_id=None, + ) + + records = db.get_all_records(table_path) + assert records is not None + assert records.schema.field("record_id_hash").type == pa.large_binary() + + def test_record_id_hash_bytes_stored_correctly(self): + """The actual bytes are stored and retrievable unchanged.""" + import pyarrow as pa + from orcapod.databases import InMemoryArrowDatabase + from orcapod.side_effects import _write_invocation_row + from orcapod.types import ContentHash + + db = InMemoryArrowDatabase() + table_path = ("test", "invocation_log2") + record_id = b"\xcd" * 16 + ch = ContentHash("sha256", bytes(range(32))) + record_id_hash_bytes = ch.to_prefixed_digest() + + _write_invocation_row( + pipeline_database=db, + table_path=table_path, + record_id=record_id, + record_id_hash_bytes=record_id_hash_bytes, + run_id="run-42", + ) + + records = db.get_all_records(table_path) + assert records is not None + stored = records.column("record_id_hash")[0].as_py() + assert isinstance(stored, bytes) + assert ContentHash.from_prefixed_digest(stored) == ch From 5bf2e3a6df624df9bbe6ca16688285965b616d38 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 02:23:45 +0000 Subject: [PATCH 8/9] docs(design_issues): note OperatorJobNode migration gap for NODE_CONTENT_HASH_COL (ITL-539) --- DESIGN_ISSUES.md | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/DESIGN_ISSUES.md b/DESIGN_ISSUES.md index 856cec27..b6a43993 100644 --- a/DESIGN_ISSUES.md +++ b/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. From 8c691843a41d97474b936ba9a5a3ee2ecc8a71a9 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 03:45:44 +0000 Subject: [PATCH 9/9] fix: improve type-mismatch error in _filter_by_content_hash, widen additional_constraints to Any (ITL-539) --- src/orcapod/core/nodes/operator_node.py | 9 +++++++++ src/orcapod/core/result_cache.py | 10 ++++++---- 2 files changed, 15 insertions(+), 4 deletions(-) diff --git a/src/orcapod/core/nodes/operator_node.py b/src/orcapod/core/nodes/operator_node.py index f5cf09ba..9510a9b8 100644 --- a/src/orcapod/core/nodes/operator_node.py +++ b/src/orcapod/core/nodes/operator_node.py @@ -775,6 +775,15 @@ 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." ) + 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) diff --git a/src/orcapod/core/result_cache.py b/src/orcapod/core/result_cache.py index 91fca68f..cef4c43b 100644 --- a/src/orcapod/core/result_cache.py +++ b/src/orcapod/core/result_cache.py @@ -10,7 +10,7 @@ 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 @@ -154,7 +154,7 @@ def _ensure_rdb_schema(self) -> None: def lookup( self, input_data: DataProtocol, - additional_constraints: dict[str, bytes] | None = None, + additional_constraints: dict[str, Any] | None = None, ) -> DataProtocol | None: """Look up a cached output data for *input_data*. @@ -168,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`` @@ -180,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: