Merge branch 'main' into eywalker/itl-563-empty-failed-outputs-lose-tag-nullability-and-abort

Author: eywalker

DESIGN_ISSUES.md

@@ -384,6 +384,36 @@ which column groups (meta, source, system_tags) are returned.
 
 ---
 
+### F15 — `FunctionPod` has no pod-level error policy; sync and async paths are inconsistent
+**Status:** open
+**Severity:** high
+**Issue:** ITL-527
+
+`FunctionPod` has no `on_error` configuration. The two execution paths behave differently:
+
+- **Sync path** (`FunctionPodStream._iter_data_*`): no exception handling at all — exceptions
+  propagate unconditionally out of the iterator.
+- **Async path** (`_FunctionPodBase.async_execute()`): exceptions are caught, the observer is
+  notified via `on_data_crash()`, and the item is **silently dropped from output** with no
+  indication to the caller and no way to configure this behaviour.
+
+The node-level `FunctionJobNode.execute()` adds `error_policy: Literal["continue", "fail_fast"]`,
+but this is only available for DB-backed execution and uses different vocabulary from the rest
+of the framework.
+
+`SinkPod` and `TapPod` (ITL-524/ITL-525) introduced a clean `on_error: Literal["raise", "log"]`
+vocabulary at the pod level. `FunctionPod` should adopt the same model: pod-level `on_error`
+config, consistent behaviour between sync and async paths, and the same `"raise"` / `"log"`
+vocabulary.
+
+**Fix needed:** Add `on_error: Literal["raise", "log"] = "raise"` to `FunctionPodConfig` (or
+equivalent). In the async path, replace the unconditional silent-drop with the configured
+behaviour: `"raise"` propagates the exception; `"log"` logs at `WARNING` and drops the item
+(current behaviour, but now explicit and configurable). Align `FunctionJobNode.error_policy`
+to use the same vocabulary.
+
+---
+
 ## `src/orcapod/core/cached_function_pod.py` / `src/orcapod/core/data_function.py`
 
 ### CFP1 — Extract shared result caching logic from CachedDataFunction and CachedFunctionPod
@@ -1291,3 +1321,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

@@ -776,7 +776,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)
 
@@ -802,7 +811,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

@@ -264,7 +264,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)
@@ -284,7 +284,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.
@@ -298,15 +298,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()