feat(pyamber): add workflow transport runtime

amber/LICENSE-binary-python

@@ -290,8 +290,8 @@ Dependencies under the BSD 3-Clause License
 
 Python packages:
   - cached-property==2.0.1
-  - click==8.4.2
   - cloudpickle==3.1.2
+  - click==8.4.2
   - contourpy==1.3.3
   - cycler==0.12.1
   - fsspec==2026.6.0

amber/requirements.txt

@@ -25,6 +25,7 @@ overrides==7.7.0
 typing_extensions==4.14.1
 bidict==0.22.0
 cached_property==2.0.1
+cloudpickle==3.1.2
 psutil==7.2.2
 tzlocal==2.1
 # Not imported directly: s3fs (with aiobotocore) backs pyiceberg's

amber/src/main/python/pytexera/workflow/__init__.py

@@ -0,0 +1,23 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Minimal keyed workflow runtime used by generated Python UDFs."""
+
+from pytexera.workflow.operators import TupleOperator
+from pytexera.workflow.runtime import Heap, InputPort, Runtime
+
+__all__ = ["Heap", "InputPort", "Runtime", "TupleOperator"]

amber/src/main/python/pytexera/workflow/codec.py

@@ -0,0 +1,130 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Cloudpickle transport for explicit workflow boundary payloads."""
+
+from __future__ import annotations
+
+import pickle
+from dataclasses import dataclass
+
+import cloudpickle
+
+
+@dataclass(frozen=True, order=True)
+class BoundaryPayload:
+    """One boundary contract, its path-present fields, and encoded values."""
+
+    boundary_id: str
+    fields: tuple[str, ...]
+    present: tuple[str, ...]
+    payload: bytes
+
+    def __post_init__(self) -> None:
+        if not self.boundary_id:
+            raise ValueError("boundary ID must be nonempty")
+        if self.fields != tuple(sorted(set(self.fields))) or any(
+            not field.isidentifier() for field in self.fields
+        ):
+            raise ValueError("boundary fields must be canonical Python names")
+        if self.present != tuple(
+            field for field in self.fields if field in frozenset(self.present)
+        ):
+            raise ValueError("present fields must be a canonical contract subset")
+        if not isinstance(self.payload, bytes):
+            raise TypeError("boundary payload must be bytes")
+
+
+@dataclass(frozen=True)
+class WorkflowEnvelope:
+    """All selected boundary payloads for one independent execution key."""
+
+    execution_key: str
+    boundaries: tuple[BoundaryPayload, ...]
+
+    def __post_init__(self) -> None:
+        if not self.execution_key:
+            raise ValueError("execution key must be nonempty")
+        if not isinstance(self.boundaries, tuple) or any(
+            not isinstance(row, BoundaryPayload) for row in self.boundaries
+        ):
+            raise TypeError("workflow boundaries must be a typed tuple")
+        ids = tuple(row.boundary_id for row in self.boundaries)
+        if ids != tuple(sorted(set(ids))):
+            raise ValueError("workflow boundaries must be canonical and unique")
+
+
+def encode_boundary(
+    boundary_id: str,
+    fields: tuple[str, ...],
+    values: tuple[object, ...],
+    *,
+    present: tuple[str, ...] | None = None,
+) -> BoundaryPayload:
+    """Encode values present on this path under one selected field contract."""
+
+    present = fields if present is None else present
+    if len(present) != len(values):
+        raise ValueError("present boundary fields and values must have equal length")
+    payload = cloudpickle.dumps(values, protocol=pickle.HIGHEST_PROTOCOL)
+    return BoundaryPayload(boundary_id, fields, present, payload)
+
+
+def decode_boundary(
+    boundary: BoundaryPayload,
+    fields: tuple[str, ...],
+) -> tuple[object, ...]:
+    """Decode a payload only under its exact selected field contract."""
+
+    if boundary.fields != fields:
+        raise ValueError("boundary fields do not match the requested contract")
+    values = cloudpickle.loads(boundary.payload)
+    if not isinstance(values, tuple) or len(values) != len(boundary.present):
+        raise ValueError("decoded boundary payload has an invalid shape")
+    return values
+
+
+def dumps_envelope(envelope: WorkflowEnvelope) -> bytes:
+    """Validate and encode a workflow envelope."""
+
+    if not isinstance(envelope, WorkflowEnvelope):
+        raise TypeError("envelope codec requires WorkflowEnvelope")
+    return cloudpickle.dumps(envelope, protocol=pickle.HIGHEST_PROTOCOL)
+
+
+def loads_envelope(payload: bytes) -> WorkflowEnvelope:
+    """Decode and type-check one workflow envelope."""
+
+    envelope = cloudpickle.loads(payload)
+    if not isinstance(envelope, WorkflowEnvelope):
+        raise TypeError("decoded payload is not WorkflowEnvelope")
+    return envelope
+
+
+def merge_envelopes(
+    left: WorkflowEnvelope,
+    right: WorkflowEnvelope,
+) -> WorkflowEnvelope:
+    """Merge independent fan-in payloads for the same execution key."""
+
+    if left.execution_key != right.execution_key:
+        raise ValueError("cannot merge envelopes with different execution keys")
+    boundaries = (*left.boundaries, *right.boundaries)
+    return WorkflowEnvelope(
+        left.execution_key,
+        tuple(sorted(boundaries, key=lambda row: row.boundary_id)),
+    )

amber/src/main/python/pytexera/workflow/operators.py

@@ -0,0 +1,147 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Generic keyed fan-in adapter between Amber tuples and workflow Runtime."""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from dataclasses import dataclass
+
+from pytexera.udf.udf_operator import UDFOperatorV2
+from pytexera.workflow.codec import (
+    WorkflowEnvelope,
+    dumps_envelope,
+    loads_envelope,
+    merge_envelopes,
+)
+from pytexera.workflow.runtime import Runtime
+
+ENVELOPE_FIELD = "workflow_envelope"
+EXECUTION_KEY_FIELD = "execution_key"
+
+
+@dataclass(frozen=True)
+class _PendingExecution:
+    envelope: WorkflowEnvelope
+    arrived_ports: frozenset[int]
+
+
+class TupleOperator(UDFOperatorV2):
+    """Execute once after all configured boundaries for one key have arrived."""
+
+    runtime: Runtime
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._pending: dict[str, _PendingExecution] = {}
+        self._finished_ports: set[int] = set()
+
+    def process_tuple(self, tuple_, port: int) -> Iterator[dict[str, object]]:
+        """Join one keyed port arrival and execute when all ports are present."""
+
+        if not self.runtime.input_ports:
+            key = _field(tuple_, EXECUTION_KEY_FIELD, "default")
+            yield _output(self.runtime.execute(WorkflowEnvelope(str(key), ())))
+            return
+        incoming = self._incoming_envelope(tuple_, port)
+        pending = self._merge_pending(incoming, port)
+        if len(pending.arrived_ports) != len(self.runtime.input_ports):
+            return
+        del self._pending[incoming.execution_key]
+        yield _output(self.runtime.execute(pending.envelope))
+
+    def _incoming_envelope(self, tuple_, port: int) -> WorkflowEnvelope:
+        """Validate and retain only boundaries owned by one physical input port."""
+
+        _validate_port(port, len(self.runtime.input_ports))
+        if port in self._finished_ports:
+            raise RuntimeError(f"input arrived after port {port} finished")
+        raw = _field(tuple_, ENVELOPE_FIELD)
+        envelope = loads_envelope(raw)
+        relevant = tuple(
+            row
+            for row in envelope.boundaries
+            if row.boundary_id in self.runtime.incoming
+        )
+        expected = frozenset(self.runtime.input_ports[port].boundaries)
+        found = frozenset(row.boundary_id for row in relevant)
+        if found != expected:
+            raise ValueError(
+                f"port {port} boundaries differ from its Runtime contract: "
+                f"expected {tuple(sorted(expected))!r}, found {tuple(sorted(found))!r}"
+            )
+        return WorkflowEnvelope(envelope.execution_key, relevant)
+
+    def _merge_pending(
+        self,
+        incoming: WorkflowEnvelope,
+        port: int,
+    ) -> _PendingExecution:
+        """Merge one validated arrival into its execution-key accumulator."""
+
+        execution_key = incoming.execution_key
+        previous = self._pending.get(execution_key)
+        if previous is not None and port in previous.arrived_ports:
+            raise RuntimeError(
+                f"execution {execution_key!r} arrived twice on port {port}"
+            )
+        merged = (
+            incoming
+            if previous is None
+            else merge_envelopes(previous.envelope, incoming)
+        )
+        arrived = (
+            frozenset((port,)) if previous is None else previous.arrived_ports | {port}
+        )
+        pending = _PendingExecution(merged, arrived)
+        self._pending[execution_key] = pending
+        return pending
+
+    def on_finish(self, port: int) -> Iterator[None]:
+        """Close one input port and reject incomplete keyed executions."""
+
+        if not self.runtime.input_ports:
+            return
+            yield None
+        _validate_port(port, len(self.runtime.input_ports))
+        if port in self._finished_ports:
+            raise RuntimeError(f"input port {port} finished twice")
+        self._finished_ports.add(port)
+        if len(self._finished_ports) == len(self.runtime.input_ports) and self._pending:
+            keys = tuple(sorted(self._pending))
+            raise RuntimeError(f"incomplete workflow executions at finish: {keys!r}")
+        return
+        yield None
+
+
+def _validate_port(port: int, count: int) -> None:
+    if not isinstance(port, int) or isinstance(port, bool) or not 0 <= port < count:
+        raise ValueError(f"input port must be an integer in [0, {count})")
+
+
+def _field(tuple_, name: str, default=...) -> object:
+    try:
+        return tuple_[name]
+    except (KeyError, IndexError):
+        if default is ...:
+            raise ValueError(f"input tuple is missing field {name!r}") from None
+        return default
+
+
+def _output(envelope: WorkflowEnvelope) -> dict[str, object]:
+    return {ENVELOPE_FIELD: dumps_envelope(envelope)}