This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new ed16a605da refactor(amber): read the loop input table from its
materialization instead of shipping it in state (#6971)
ed16a605da is described below
commit ed16a605dacc0185bc5f069d14157ee3e128b68a
Author: Xinyuan Lin <[email protected]>
AuthorDate: Mon Aug 10 23:57:45 2026 -0700
refactor(amber): read the loop input table from its materialization instead
of shipping it in state (#6971)
### What changes were proposed in this PR?
The loop's input table used to ride **inside the State content**:
LoopStart encoded its buffered input as Arrow IPC bytes, base64'd into
the JSON `content` column, and that payload was re-written and re-read
at **every loop-body hop, every iteration**.
That data already exists. In the fully-materialized mode loops require,
the Loop Start's input-port materialization holds exactly the loop's
input table for the whole loop — Loop Start re-reads it every iteration,
and the back-edge truncates only the *state* doc at the same base URI,
never the *result* doc. So this PR ships the port's **base URI** in the
setup config and derives both addresses from it:
```
loopStartPortUris[LoopStart-id] = <base URI of LoopStart's input port>
├── state_uri(base) → back-edge write address (as before,
derived)
└── result_uri(base) → the loop's input table (NEW: read at
EndChannel)
```
| Piece | Before | After |
|---|---|---|
| proto field 4 | `loopStartStateUris` = state URI | `loopStartPortUris`
= base URI (renamed so the semantic change is loud) |
| LoopStart's produced state | user vars + IPC-encoded table (base64 in
JSON) | user vars only — small, pure JSON |
| Loop-body hops | fat state re-materialized per hop, per iteration |
tiny state |
| LoopEnd's table | decoded from state content | read once per iteration
from `result_uri(base)` at EndChannel, injected via a new runtime-only
`attach_loop_table` hook |
| `table_to_ipc_bytes` / `table_from_ipc_bytes` | second, divergent
Arrow codec (lossy `from_pandas` inference) | deleted — the read goes
through the canonical iceberg reader |
**When the read happens matters.** The read is issued at **EndChannel**
(the matching state is stashed at consume and the operator's update runs
in `complete()`), not at consume time. At consume time this worker's own
materialization reader is still streaming, and issuing a second
iceberg/S3 read from the main loop thread in that window made the reader
fail with S3 `Access Denied` — `LoopIntegrationSpec` hung to the CI job
timeout on both OSes. Deferring past the reader removes the overlap:
integration went from a 20-minute cancel to green in ~9 minutes. The
matching consume emits no state downstream, so moving it is unobservable
outside the operator.
Semantics deliberately preserved:
- the reserved-`table` collision raise stays (a user var named `table`
would now be *silently shadowed* by the injected table — worse than
before);
- the "consumed" marker (`_loop_table`) is still set only by a
**successful** `run_update`, so `condition()`'s short-circuit for
pass-through-only Loop Ends is unchanged;
- nested loops work by construction: the inner Loop Start's entry points
at the outer Loop Start's output port, whose result doc is recreated per
*outer* iteration but persists across *inner* iterations (the jump
rewinds to the inner level only).
Wins: no ~33% base64 bloat, no JSON-column size ceiling on the table
(large-table loops become viable), strictly less I/O for any non-empty
loop body (one read per iteration replaces N state-doc writes+reads per
hop), and one Arrow codec instead of two.
Note: this deepens the read-side use of `storagePairs.head._1` — the
same shared upstream URI as the known back-edge fan-out design
discussion; if that ever moves to a per-loop private doc, this read
moves with it.
### Any related issues, documentation, discussions?
Builds on #5900 (State columns) and #6661 (envelope through JVM hops).
Related design context: #6660.
### How was this PR tested?
- **Unit** — `test_loop_operators.py` rewritten for the attach-based
flow plus new pins: the produced state carries no `table`; attaching
alone does **not** mark the loop consumed; `run_update` fails loud when
no table was attached. `test_main_loop.py` pins the base-URI derivation
for the back-edge write (`state_uri(base)`), the missing-config
fail-loud, and that the matching consume stashes the state without
touching storage, with the read + update happening once at EndChannel.
`test_initialize_executor_handler.py` covers the renamed proto field.
237 tests green locally (the only failures in a full sweep are
pre-existing environment ones, identical on unmodified main).
- **Scala** — full test-compile (proto regen included),
`scalafmtCheckAll`, `scalafixAll --check`, and the worker/descriptor
spec suites (`WorkerSpec`, `WorkflowWorkerSpec`,
`SerializationManagerSpec`, `WorkflowExecutionManagerSpec`,
`LoopStartOpDescSpec`, `LoopEndOpDescSpec`) all pass on Java 17.
- **E2E** — the four `LoopIntegrationSpec` cases (single, nested 3×3,
JVM chain, nested JVM chain) exercise the full read path in the
`amber-integration` CI job (both jobs green in ~9 min); the nested cases
specifically cover the inner-loop read against the outer Loop Start's
per-outer-iteration output doc.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Fable 5)
---
.../texera/amber/bench/ArrowFlightActorBench.scala | 2 +-
.../engine/architecture/rpc/controlcommands.proto | 21 +-
.../control/initialize_executor_handler.py | 4 +-
.../python/core/architecture/managers/context.py | 6 +-
amber/src/main/python/core/models/operator.py | 89 +++---
amber/src/main/python/core/models/table.py | 29 --
amber/src/main/python/core/runnables/main_loop.py | 198 ++++++++++--
.../core/util/console_message/replace_print.py | 13 +-
.../scheduling/RegionExecutionManager.scala | 10 +-
.../scheduling/WorkflowExecutionManager.scala | 19 +-
.../amber/engine/e2e/LoopIntegrationSpec.scala | 12 +-
.../control/test_initialize_executor_handler.py | 16 +-
.../test/python/core/models/test_loop_operators.py | 151 ++++-----
.../test/python/core/runnables/test_main_loop.py | 345 ++++++++++++++++++---
.../engine/architecture/worker/WorkerSpec.scala | 2 +-
.../architecture/worker/WorkflowWorkerSpec.scala | 2 +-
.../worker/managers/SerializationManagerSpec.scala | 2 +-
.../texera/amber/core/workflow/PhysicalOp.scala | 5 +-
.../amber/operator/loop/LoopEndOpDescSpec.scala | 11 +-
.../amber/operator/loop/LoopStartOpDescSpec.scala | 2 +-
20 files changed, 660 insertions(+), 279 deletions(-)
diff --git
a/amber/src/bench/scala/org/apache/texera/amber/bench/ArrowFlightActorBench.scala
b/amber/src/bench/scala/org/apache/texera/amber/bench/ArrowFlightActorBench.scala
index e3b79ea804..602f4f0841 100644
---
a/amber/src/bench/scala/org/apache/texera/amber/bench/ArrowFlightActorBench.scala
+++
b/amber/src/bench/scala/org/apache/texera/amber/bench/ArrowFlightActorBench.scala
@@ -287,7 +287,7 @@ object ArrowFlightActorBench {
1,
OpExecWithCode(IdentityPythonCode, "python"),
isSource = false,
- loopStartStateUris = Map.empty
+ loopStartPortUris = Map.empty
),
ctx,
0L
diff --git
a/amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/rpc/controlcommands.proto
b/amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/rpc/controlcommands.proto
index 8a6403a97e..e60b7cd74f 100644
---
a/amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/rpc/controlcommands.proto
+++
b/amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/rpc/controlcommands.proto
@@ -254,12 +254,21 @@ message InitializeExecutorRequest {
int32 totalWorkerCount = 1;
core.OpExecInitInfo opExecInitInfo = 2;
bool isSource = 3;
- // Loop-back write addresses: Loop Start logical operator id -> state URI of
- // that Loop Start's input port. Constant per execution (minted at schedule
- // time); a Loop End worker selects the entry by the loop_start_id carried on
- // the consumed StateFrame and writes the next-iteration state there. Empty
- // for plans without loops.
- map<string, string> loopStartStateUris = 4;
+ // Loop bookkeeping addresses: Loop Start logical operator id -> the BASE
+ // URI of the materialized port that Loop Start's input port READS FROM,
+ // i.e. the upstream operator's output-port materialization (the storage
+ // pair of the Loop Start's input port config), not a port of the Loop
+ // Start itself. Constant per execution
+ // (minted at schedule time); a Loop End worker selects the entry by the
+ // loop_start_id carried on the consumed StateFrame and derives two URIs
+ // from the base:
+ // - state URI (VFSURIFactory.state_uri): the loop-back write address for
+ // the next-iteration state;
+ // - result URI (VFSURIFactory.result_uri): the loop's input table, read
+ // at consume time so the table never rides inside the State content
+ // through the loop body.
+ // Empty for plans without loops.
+ map<string, string> loopStartPortUris = 4;
}
message UpdateExecutorRequest {
diff --git
a/amber/src/main/python/core/architecture/handlers/control/initialize_executor_handler.py
b/amber/src/main/python/core/architecture/handlers/control/initialize_executor_handler.py
index 50fceab7fe..d98e67bcce 100644
---
a/amber/src/main/python/core/architecture/handlers/control/initialize_executor_handler.py
+++
b/amber/src/main/python/core/architecture/handlers/control/initialize_executor_handler.py
@@ -30,6 +30,6 @@ class InitializeExecutorHandler(ControlHandler):
self.context.executor_manager.initialize_executor(
op_exec_with_code.code, req.is_source, op_exec_with_code.language
)
- # Loop-back write addresses; see the proto field doc on
loopStartStateUris.
- self.context.loop_start_state_uris = dict(req.loop_start_state_uris)
+ # Loop bookkeeping base URIs; see the proto field doc on
loopStartPortUris.
+ self.context.loop_start_port_uris = dict(req.loop_start_port_uris)
return EmptyReturn()
diff --git a/amber/src/main/python/core/architecture/managers/context.py
b/amber/src/main/python/core/architecture/managers/context.py
index dfcc30f9aa..18bbdd3958 100644
--- a/amber/src/main/python/core/architecture/managers/context.py
+++ b/amber/src/main/python/core/architecture/managers/context.py
@@ -87,9 +87,9 @@ class Context:
self.debug_manager = DebugManager(
self.tuple_processing_manager.context_switch_condition
)
- # Loop-back write addresses delivered at setup; see the proto field doc
- # on InitializeExecutorRequest.loopStartStateUris
(controlcommands.proto).
- self.loop_start_state_uris: Dict[str, str] = {}
+ # Loop bookkeeping base URIs delivered at setup; see the proto field
doc
+ # on InitializeExecutorRequest.loopStartPortUris
(controlcommands.proto).
+ self.loop_start_port_uris: Dict[str, str] = {}
def report_exception(self, err: BaseException) -> None:
"""Route an operator-facing exception to the exception manager and
diff --git a/amber/src/main/python/core/models/operator.py
b/amber/src/main/python/core/models/operator.py
index c27dc154b5..682b18bb64 100644
--- a/amber/src/main/python/core/models/operator.py
+++ b/amber/src/main/python/core/models/operator.py
@@ -24,7 +24,7 @@ from typing import Iterator, List, Mapping, Optional, Union,
MutableMapping, Pro
from . import Table, TableLike, Tuple, TupleLike, Batch, BatchLike
from .state import State
-from .table import all_output_to_tuple, table_from_ipc_bytes,
table_to_ipc_bytes
+from .table import all_output_to_tuple
import base64
@@ -279,17 +279,6 @@ class TableOperator(TupleOperatorV2):
table = Table(self.__table_data[port])
yield from self.process_table(table, port)
- def _buffered_table(self, port: int) -> Table:
- """Tuples buffered for ``port`` so far, materialized as a Table.
-
- Exposed so subclasses (e.g. ``LoopStartOperator``) can read the
- buffer outside the ``process_table`` callback without reaching into
- the parent's name-mangled private field. Inside this class
- ``self.__table_data`` resolves via normal name mangling, so a future
- rename of ``TableOperator`` keeps callers transparent.
- """
- return Table(self.__table_data[port])
-
@abstractmethod
def process_table(self, table: Table, port: int) ->
Iterator[Optional[TableLike]]:
"""
@@ -308,10 +297,13 @@ class TableOperator(TupleOperatorV2):
# namespaces the loop expressions run in. It is NOT user state: a user loop
# variable of the same name collides with it, so both operators raise on
# collision (see ``_reserved_name_error``) rather than silently dropping the
-# user's value. The envelope names (``loop_counter`` / ``loop_start_id``) never
-# enter user state -- they ride the StateFrame envelope (see
-# ``core.models.payload``). The loop-back write address is setup config, not
-# state (see ``loopStartStateUris`` on the ``InitializeExecutorRequest``
proto).
+# user's value. The table itself never rides the State content: the LoopEnd
+# runtime reads it from the Loop Start's input-port materialization at consume
+# time and injects it via ``attach_loop_table``. The envelope names
+# (``loop_counter`` / ``loop_start_id``) never enter user state -- they ride
+# the StateFrame envelope (see ``core.models.payload``). The loop bookkeeping
+# base URI is setup config, not state (see ``loopStartPortUris`` on the
+# ``InitializeExecutorRequest`` proto).
_TABLE_KEY = "table"
_RESERVED_STATE_KEYS: frozenset = frozenset({_TABLE_KEY})
@@ -359,8 +351,9 @@ class LoopStartOperator(TableOperator):
``open()`` seeds ``self.state`` with the user's loop variables;
``process_state`` merges upstream state in; ``produce_state_on_finish``
- emits those variables plus the input table (Arrow IPC; see
- ``table_to_ipc_bytes`` in ``core.models.table``) to the matching LoopEnd.
+ emits those variables to the matching LoopEnd. The input table does NOT
+ ride the state: the LoopEnd runtime reads it from this operator's
+ input-port materialization at consume time (``loopStartPortUris``).
``loop_counter`` and the nested pass-through are owned by
``MainLoop._process_state_frame``, not this operator.
@@ -404,18 +397,15 @@ class LoopStartOperator(TableOperator):
@overrides.final
def produce_state_on_finish(self, port: int) -> State:
- # Emit the user's loop variables plus the buffered input table for the
- # matching LoopEnd. The table rides as an Arrow IPC stream, not pickle
- # (see `table_to_ipc_bytes` in core.models.table for why). Reads the
- # buffer through `_buffered_table` so a rename of `TableOperator`
- # doesn't silently break this.
- # A user loop variable named `table` would be overwritten by the input
- # table below, so flag the collision instead of silently dropping it.
+ # Emit the user's loop variables for the matching LoopEnd. The input
+ # table does NOT ride the state: the LoopEnd runtime reads it from
+ # this operator's input-port materialization at consume time
+ # (loopStartPortUris). A user loop variable named `table` would be
+ # silently shadowed by that injected table in the LoopEnd's
+ # update/condition namespaces, so flag the collision here.
if _TABLE_KEY in self.state:
raise _reserved_name_error(_TABLE_KEY)
- produced = State(self.state)
- produced[_TABLE_KEY] = table_to_ipc_bytes(self._buffered_table(port))
- return produced
+ return State(self.state)
class LoopEndOperator(TableOperator):
@@ -428,9 +418,10 @@ class LoopEndOperator(TableOperator):
``eval_condition``); all substantive logic lives here.
``process_table`` yields each input table through as-is; ``process_state``
- runs the user's ``update`` and persists only user variables back into
- ``self.state`` (keeping the decoded table on ``self._loop_table``);
- ``condition()`` decides whether ``MainLoop.complete()`` fires the
back-edge.
+ runs the user's ``update`` (against the table the runtime attached via
+ ``attach_loop_table``) and persists only user variables back into
+ ``self.state``; ``condition()`` decides whether ``MainLoop.complete()``
+ fires the back-edge.
Subclass contract: the generated subclass overrides ``process_state()`` and
``condition()`` only; all other methods are ``@overrides.final``.
@@ -452,21 +443,47 @@ class LoopEndOperator(TableOperator):
# AttributeError; a None _loop_table means "nothing consumed yet" and
# condition() short-circuits to False (see eval_condition).
self.state: State = State()
+ # Set by the runtime (attach_loop_table) right before the matching
+ # consume and taken (cleared) by run_update. The clear is defensive
+ # rather than load-bearing today -- a worker instance handles a single
+ # iteration (workers are recreated per region execution), so there is
+ # no second consume within one instance for a stale table to leak
+ # into -- but nothing may run an update against a table it was not
+ # explicitly handed. It stays separate from _loop_table because
+ # _loop_table doubles as the "consumed" marker that condition()
+ # short-circuits on, and only a SUCCESSFUL update may set that.
+ self._attached_table: Optional[Table] = None
self._loop_table: Optional[Table] = None
@overrides.final
def process_table(self, table: Table, port: int) ->
Iterator[Optional[TableLike]]:
yield table
+ @overrides.final
+ def attach_loop_table(self, table: Table) -> None:
+ # Runtime-only hook: MainLoop reads the loop's input table from the
+ # Loop Start's input-port materialization (loopStartPortUris) and
+ # attaches it here right before the matching consume, so the table
+ # never has to ride inside the State content through the loop body.
+ self._attached_table = table
+
@overrides.final
def run_update(self, update_code: str, state: State) -> None:
# Run the user's `update` in a throwaway namespace seeded with the
# incoming loop variables and the input table, then persist the user
- # variables back into self.state. The table arrives as an Arrow IPC
- # stream, not pickle (see `table_to_ipc_bytes` in core.models.table
- # for why); the decoded table is kept on self._loop_table so
- # condition() can read it after the update.
- input_table = table_from_ipc_bytes(state[_TABLE_KEY])
+ # variables back into self.state. The table is attached by the runtime
+ # (attach_loop_table) from the Loop Start's input materialization; on
+ # a successful update it is kept on self._loop_table so condition()
+ # can read it afterwards.
+ if self._attached_table is None:
+ raise RuntimeError(
+ "loop input table was not attached before the update; the "
+ "runtime must call attach_loop_table on the matching consume"
+ )
+ input_table = self._attached_table
+ # Take it: a later iteration that never got a table must raise above
+ # rather than silently run the update against the previous one.
+ self._attached_table = None
namespace = {**state, _TABLE_KEY: input_table}
# Pass the namespace as exec globals (not a locals-only mapping) so a
# comprehension / generator expression / lambda in the user's `update`
diff --git a/amber/src/main/python/core/models/table.py
b/amber/src/main/python/core/models/table.py
index a85e75cbef..4716e0eba0 100644
--- a/amber/src/main/python/core/models/table.py
+++ b/amber/src/main/python/core/models/table.py
@@ -16,7 +16,6 @@
# under the License.
import pandas
-import pyarrow as pa
from pampy import match
from typing import Iterator, TypeVar, List
@@ -83,34 +82,6 @@ class Table(pandas.DataFrame):
return super().__eq__(other).all()
-def table_to_ipc_bytes(table: Table) -> bytes:
- """Serialize ``table`` as an Apache Arrow IPC stream.
-
- Used by the loop operators to round-trip a Table through a state dict
- (and through iceberg storage) without resorting to ``pickle.dumps``,
- which would expose ``pickle.loads`` as a remote-code-execution surface
- on the receiving side. Arrow IPC is a length-prefixed, schema-typed
- format that carries data only -- no executable payload.
- """
- arrow_table = pa.Table.from_pandas(table, preserve_index=False)
- sink = pa.BufferOutputStream()
- with pa.ipc.new_stream(sink, arrow_table.schema) as writer:
- writer.write_table(arrow_table)
- return sink.getvalue().to_pybytes()
-
-
-def table_from_ipc_bytes(buf: bytes) -> Table:
- """Inverse of :func:`table_to_ipc_bytes`.
-
- Reconstruct a Table from an Apache Arrow IPC stream buffer. Raises if
- ``buf`` is not a well-formed Arrow IPC stream, so malformed input
- surfaces as a parse error rather than executing anything.
- """
- with pa.ipc.open_stream(pa.py_buffer(buf)) as reader:
- arrow_table = reader.read_all()
- return Table(arrow_table.to_pandas())
-
-
def all_output_to_tuple(output) -> Iterator[Tuple]:
"""
Convert all kinds of types into Tuples.
diff --git a/amber/src/main/python/core/runnables/main_loop.py
b/amber/src/main/python/core/runnables/main_loop.py
index 048bc0a289..108078ba4f 100644
--- a/amber/src/main/python/core/runnables/main_loop.py
+++ b/amber/src/main/python/core/runnables/main_loop.py
@@ -30,6 +30,7 @@ from core.architecture.rpc.async_rpc_server import
AsyncRPCServer
from core.models import (
InternalQueue,
StateFrame,
+ Table,
Tuple,
)
from core.models.internal_marker import StartChannel, EndChannel
@@ -43,7 +44,9 @@ from core.models.operator import LoopEndOperator,
LoopStartOperator
from core.models.state import State
from core.runnables.data_processor import DataProcessor
from core.storage.document_factory import DocumentFactory
+from core.storage.vfs_uri_factory import VFSURIFactory
from core.util import StoppableQueueBlockingRunnable, get_one_of
+from core.util.console_message.replace_print import replace_print
from core.util.console_message.timestamp import current_time_in_local_timezone
from core.util.customized_queue.queue_base import QueueElement
from core.util.virtual_identity import get_logical_op_id
@@ -85,19 +88,26 @@ class MainLoop(StoppableQueueBlockingRunnable):
self._output_queue: InternalQueue = output_queue
# Captured from the consumed StateFrame envelope when a matching
# LoopEnd (loop_counter == 0) takes a state; used for the jump RPC
- # and the setup-config URI lookup (context.loop_start_state_uris).
+ # and the setup-config URI lookup (context.loop_start_port_uris).
self._loop_start_id: str = ""
- # Whether this LoopEnd already consumed its loop state this execution.
+ # Whether this LoopEnd already took its loop state this execution.
# A loop body may branch and converge on the Loop End, and every reader
# on its input port replays that branch's states independently, so the
- # same iteration's state arrives once per branch. Workers are recreated
- # on each region re-execution, so this instance flag is per iteration.
+ # same iteration's state arrives once per branch. Cleared when the
+ # deferred consume takes the stash (_consume_pending_loop_state), so
+ # the flag is per-execution by construction -- not by the scheduler
+ # happening to recreate workers each iteration.
self._loop_state_consumed: bool = False
# Whether this LoopEnd forwarded an UNstamped counter-0 state instead
# of consuming it. Paired with _loop_start_id by
# _check_loop_state_arrived: forwarding one and never capturing a stamp
# means the loop's own state reached here with its stamp lost.
self._forwarded_unstamped_state: bool = False
+ # The matching state a LoopEnd will consume, stashed when taken and
+ # actually handed to the operator at EndChannel (see
+ # _consume_pending_loop_state for why the table read must not overlap
+ # this worker's own materialization reader).
+ self._pending_loop_state: Optional[State] = None
self.context = Context(worker_id, input_queue)
self._async_rpc_server = AsyncRPCServer(output_queue,
context=self.context)
@@ -108,21 +118,98 @@ class MainLoop(StoppableQueueBlockingRunnable):
target=self.data_processor.run, daemon=True,
name="data_processor_thread"
).start()
- def _jump_to_loop_start(
- self, executor: LoopEndOperator, coordinator_interface
- ) -> None:
- # The write address is setup config, keyed by the captured id. Fail
- # loud BEFORE the jump RPC so a misconfigured loop does not rewind the
- # schedule without a back-edge write. Anything raised here (a missing
- # URI, or a failed state write after the jump) is reported by
- # complete()'s guard as an operator-facing error.
- uri = self.context.loop_start_state_uris.get(self._loop_start_id)
+ def _loop_start_base_uri(self) -> str:
+ # The loop's bookkeeping base URI is setup config, keyed by the
+ # captured id (see InitializeExecutorRequest.loopStartPortUris). Fail
+ # loud on a missing entry: anything raised here is reported by the
+ # caller's guard as an operator-facing error.
+ uri = self.context.loop_start_port_uris.get(self._loop_start_id)
if not uri:
raise RuntimeError(
- f"no loop-back state URI configured for LoopStart "
+ f"no loop bookkeeping URI configured for LoopStart "
f"'{self._loop_start_id}' "
- f"(have: {sorted(self.context.loop_start_state_uris)})"
+ f"(have: {sorted(self.context.loop_start_port_uris)})"
)
+ return uri
+
+ def _read_loop_input_table(self) -> Table:
+ # The loop's input table is the Loop Start's input-port
+ # materialization: the output doc of whatever feeds this loop level.
+ # It is stable for the duration of the loop level that reads it -- the
+ # back-edge rewrites only the state doc under the same base URI, never
+ # this result doc. For an INNER loop that upstream is the outer Loop
+ # Start, whose output doc is recreated on each OUTER iteration; that is
+ # still stable across every inner iteration that reads it, which is the
+ # window that matters here. Reading it means the table never has to
+ # ride inside the State content through the loop body. Callers must
+ # invoke this OUTSIDE the window where this worker's own
materialization
+ # reader is streaming -- see _consume_pending_loop_state.
+ result_uri = VFSURIFactory.result_uri(self._loop_start_base_uri())
+ document, schema = DocumentFactory.open_document(result_uri)
+ # Normalize the same way as the only other reader of this doc: the
+ # input-port reader casts every tuple to the doc's schema before it
+ # reaches an operator (input_port_materialization_reader_runnable), so
+ # this read must too -- two readers of one document normalizing
+ # differently would diverge on the first storage/provider change.
+ rows = []
+ for tup in document.get():
+ tup.cast_to_schema(schema)
+ rows.append(tup)
+ return Table(rows)
+
+ def _consume_pending_loop_state(self, executor: LoopEndOperator) -> None:
+ # Run the matching consume that _process_state_frame deferred.
+ #
+ # The loop's input table is read from the Loop Start's input-port
+ # materialization (see _read_loop_input_table for its lifetime). The
+ # read is deferred to here rather than done when the state arrives: at
+ # arrival time THIS worker's own materialization reader is still
+ # streaming its input, and in runs where this read overlapped that
+ # reader, the reader failed with S3 "Access Denied" (MinIO's answer
+ # for a deleted key) while iterating a lazily-pinned snapshot of a doc
+ # that region re-execution drops and recreates. Removing the overlap
+ # made those failures stop; that the overlap CAUSED them is the
+ # working hypothesis, not a ruled-out fact -- a doc dropped under a
+ # live reader would break the reader with or without this read -- so
+ # treat a recurrence as new evidence. By EndChannel the reader has
+ # finished, so the two never overlap.
+ if self._pending_loop_state is None:
+ return
+ pending = self._pending_loop_state
+ self._pending_loop_state = None
+ # Re-arm the duplicate guard here rather than relying on the scheduler
+ # recreating workers each iteration: EndChannel is PORT_ALIGNMENT, so
+ # nothing more can arrive on the port -- every duplicate is already in
+ # by the time this consume runs.
+ self._loop_state_consumed = False
+ executor.attach_loop_table(self._read_loop_input_table())
+ # A Loop End has exactly one input port (port 0); the generated
+ # operator's process_state ignores the port anyway. That single input
+ # port is also what makes THIS a safe place for the read above:
+ # EndChannel is PORT_ALIGNMENT (dispatched once every channel of the
+ # port has delivered it) and each reader emits its EndChannel only
+ # after its iterator is exhausted, so with one input port every reader
+ # of this worker has finished before the read. A second input port
+ # would break that -- alignment is per port -- and silently
+ # reintroduce the reader/read overlap this deferral removes.
+ # The user's `update` runs here, on the main loop thread rather than
+ # inside DataProcessor._executor_session, so capture its prints
+ # explicitly -- otherwise they go to the worker's stdout instead of
+ # the console.
+ with replace_print(
+ self.context.worker_id,
self.context.console_message_manager.print_buf
+ ):
+ executor.process_state(pending, 0)
+
+ def _jump_to_loop_start(
+ self, executor: LoopEndOperator, coordinator_interface
+ ) -> None:
+ # Resolve the write address BEFORE the jump RPC so a misconfigured
+ # loop does not rewind the schedule without a back-edge write.
+ # Anything raised here (a missing URI, or a failed state write after
+ # the jump) is reported by complete()'s guard as an operator-facing
+ # error.
+ uri = VFSURIFactory.state_uri(self._loop_start_base_uri())
coordinator_interface.jump_to_operator_region(
JumpToOperatorRegionRequest(OperatorIdentity(self._loop_start_id))
)
@@ -195,12 +282,23 @@ class MainLoop(StoppableQueueBlockingRunnable):
# worker, instead of killing the thread through run()'s
# @logger.catch(reraise=True).
try:
- if executor.condition():
+ # condition() is user code on the main loop thread, same as the
+ # `update` in the deferred consume: capture its prints too.
+ with replace_print(
+ self.context.worker_id,
+ self.context.console_message_manager.print_buf,
+ ):
+ should_iterate = executor.condition()
+ if should_iterate:
self._jump_to_loop_start(executor, coordinator_interface)
except Exception as err:
self.context.report_exception(err)
self._check_exception()
return
+ # The opening flush above happened before condition() ran, and the
+ # worker is about to shut down, so anything it printed would never
+ # be sent. (_check_exception flushes on the error path.)
+ self._check_and_report_console_messages(force_flush=True)
executor.close()
# stop the data processing thread
self.data_processor.stop()
@@ -298,7 +396,7 @@ class MainLoop(StoppableQueueBlockingRunnable):
executor = self.context.executor_manager.executor
if isinstance(executor, LoopStartOperator):
# A LoopStart stamps its own logical op id; the write address
- # is setup config
(InitializeExecutorRequest.loop_start_state_uris).
+ # is setup config
(InitializeExecutorRequest.loop_start_port_uris).
output_loop_start_id =
get_logical_op_id(self.context.worker_id)
self._emit_and_save_state(
output_state,
@@ -452,29 +550,49 @@ class MainLoop(StoppableQueueBlockingRunnable):
# A real loop state is always stamped: the matching LoopStart
# stamps its own id on every iteration's output state.
# Forward it downstream unchanged, skipping the operator, like
- # any default pass-through: consuming it would clobber the
- # captured back-jump id with "" and hand run_update a State
- # with no `table` payload.
+ # any default pass-through: taking it would clobber the
+ # captured back-jump id with "" and then fail the deferred
+ # consume's bookkeeping-URI lookup for LoopStart ''.
+ #
+ # This MUST stay ahead of the dedup and the stash below: an
+ # unstamped frame is not the loop's own state, so it may
+ # neither mark that state as taken nor be handed to the
+ # operator as the iteration's state.
self._forwarded_unstamped_state = True
self._emit_and_save_state(state, in_counter,
frame.loop_start_id)
self._check_and_process_control()
return
# Matching LoopEnd (in_counter == 0, stamped): it will consume this
# state and jump back. Remember which LoopStart to jump to (it
- # rides the envelope) for complete()/_jump_to_loop_start.
+ # rides the envelope) for complete()/_jump_to_loop_start, and STASH
+ # the state -- the operator runs its update at EndChannel instead
+ # of here, because the loop's input table is read from storage and
+ # that read must not overlap this worker's own materialization
+ # reader (see _consume_pending_loop_state). The LoopEnd emits no
+ # state downstream on the matching consume, so deferring it
+ # changes nothing observable outside the operator.
#
# With a branching loop body, each branch's reader replays the same
- # iteration's state, so this fires once per inbound link. Consume
- # only the first: running the user's `update` again would advance
- # the loop variables once per branch. The copies are identical --
- # they are the same state emitted by the one matching LoopStart --
- # so dropping them loses nothing (a consume emits nothing
- # downstream either way).
+ # iteration's state, so this fires once per inbound link. The
+ # deferred consume runs the user's update exactly once either way
+ # (and run_update seeds its namespace from the incoming copy, so
+ # re-running it on an identical copy would even be idempotent);
+ # what this flag pins is WHICH copy is stashed -- the first,
+ # instead of each later arrival silently overwriting
+ # _pending_loop_state. The copies are the same state emitted by
+ # the one matching LoopStart unless a body operator overrides
+ # process_state to forward a modified one; rewriting the loop
+ # state in a branching body is not supported (which copy arrives
+ # first is scheduling-dependent either way), so take the first
+ # deliberately.
if self._loop_state_consumed:
self._check_and_process_control()
return
self._loop_state_consumed = True
self._loop_start_id = frame.loop_start_id
+ self._pending_loop_state = state
+ self._check_and_process_control()
+ return
self.context.state_processing_manager.current_input_state = state
self.process_input_state(
@@ -498,11 +616,31 @@ class MainLoop(StoppableQueueBlockingRunnable):
self._check_exception()
if self.context.exception_manager.has_exception():
# A state-emission error was reported on the main loop thread (see
- # _emit_and_save_state). Hold the region: skip port_completed and
- # complete() so the coordinator does not mark the region complete
- # (region completion is port-based) with partial, single-iteration
+ # _emit_and_save_state). Hold the region BEFORE any loop work --
+ # no storage read and no user `update` stacked on top of an
+ # already-reported error -- and skip port_completed / complete()
+ # so the coordinator does not mark the region complete (region
+ # completion is port-based) with partial, single-iteration
# results. The reported error surfaces instead of a false success.
return
+ # Run the deferred loop consume HERE, not in complete(): complete() is
+ # the tail of this method, by which point port_completed has been sent
+ # for every output port, so a failing read or `update` would be
+ # reported after the coordinator already considers the region done
+ # (region completion is port-based) -- a reported error that still
+ # reads as success. This is still past the reader: EndChannel means it
+ # finished streaming.
+ executor = self.context.executor_manager.executor
+ if isinstance(executor, LoopEndOperator):
+ try:
+ self._consume_pending_loop_state(executor)
+ except Exception as err:
+ self.context.report_exception(err)
+ self._check_exception()
+ if self.context.exception_manager.has_exception():
+ # The deferred consume failed (the table read or the user's
+ # update). Hold the region for the same reason as above.
+ return
self.process_input_tuple()
input_port_id = self.context.input_manager.get_port_id(
diff --git a/amber/src/main/python/core/util/console_message/replace_print.py
b/amber/src/main/python/core/util/console_message/replace_print.py
index 7feeeeb52d..6a31050cd5 100644
--- a/amber/src/main/python/core/util/console_message/replace_print.py
+++ b/amber/src/main/python/core/util/console_message/replace_print.py
@@ -64,6 +64,13 @@ class replace_print(ContextManager):
if "file" in kwargs:
self.builtins_print(*args, **kwargs)
return
+ # The frame that called print(). Look __name__ up with .get, not
+ # []: code run through eval/exec against a bare dict (a Loop End's
+ # condition/update, a Loop Start's output) has no __name__ in its
+ # frame globals -- eval/exec inject __builtins__ but never
+ # __name__ -- and a print() there must not crash the capture.
+ caller = inspect.currentframe().f_back
+ module_name = caller.f_globals.get("__name__", "<unknown>")
with StringIO() as tmp_buf, redirect_stdout(tmp_buf):
self.builtins_print(*args, **kwargs)
complete_str = tmp_buf.getvalue()
@@ -71,11 +78,7 @@ class replace_print(ContextManager):
worker_id=self.worker_id,
timestamp=current_time_in_local_timezone(),
msg_type=ConsoleMessageType.PRINT,
- source=(
-
f"{inspect.currentframe().f_back.f_globals['__name__']}"
- f":{inspect.currentframe().f_back.f_code.co_name}"
- f":{inspect.currentframe().f_back.f_lineno}"
- ),
+
source=(f"{module_name}:{caller.f_code.co_name}:{caller.f_lineno}"),
title=complete_str,
message="",
)
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala
index 0b32b74101..2f19887cc3 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala
@@ -113,10 +113,10 @@ class RegionExecutionManager(
maxTerminationAttempts: Int =
RegionExecutionManager.DefaultMaxTerminationAttempts,
killRetryBaseBackoffMs: Long =
RegionExecutionManager.DefaultKillRetryBaseBackoffMs,
killRetryTimer: Timer = new JavaTimer(true),
- // Loop-back write addresses (Loop Start logical op id -> its input port's
- // state URI), shipped to every worker in InitializeExecutorRequest. See
- // WorkflowExecutionManager.loopStartStateUris.
- loopStartStateUris: Map[String, String] = Map.empty
+ // Loop bookkeeping addresses (Loop Start logical op id -> its input
+ // port's BASE materialization URI), shipped to every worker in
+ // InitializeExecutorRequest. See
WorkflowExecutionManager.loopStartPortUris.
+ loopStartPortUris: Map[String, String] = Map.empty
) extends AmberLogging {
initRegionExecution()
@@ -452,7 +452,7 @@ class RegionExecutionManager(
workerConfigs.length,
physicalOp.opExecInitInfo,
physicalOp.isSourceOperator,
- loopStartStateUris
+ loopStartPortUris
),
asyncRPCClient.mkContext(workerId)
)
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/WorkflowExecutionManager.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/WorkflowExecutionManager.scala
index 0b63d1f17c..b736cd171c 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/WorkflowExecutionManager.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/WorkflowExecutionManager.scala
@@ -21,7 +21,6 @@ package org.apache.texera.amber.engine.architecture.scheduling
import com.twitter.util.Future
import com.typesafe.scalalogging.LazyLogging
-import org.apache.texera.amber.core.storage.VFSURIFactory
import org.apache.texera.amber.core.workflow.{GlobalPortIdentity, PhysicalLink}
import
org.apache.texera.amber.engine.architecture.scheduling.config.InputPortConfig
import org.apache.texera.amber.engine.architecture.common.{
@@ -57,15 +56,21 @@ class WorkflowExecutionManager(
}
/**
- * Loop-back write addresses shipped to every worker at setup; semantics are
- * documented on `InitializeExecutorRequest.loopStartStateUris`
(controlcommands.proto).
+ * Loop bookkeeping addresses shipped to every worker at setup; semantics
are
+ * documented on `InitializeExecutorRequest.loopStartPortUris`
(controlcommands.proto).
+ * The value is the BASE URI of the materialized port the Loop Start's
+ * input port READS FROM -- the upstream operator's output-port
+ * materialization (`cfg.storagePairs.head._1` below), not a port of the
+ * Loop Start itself. The Python worker derives the state URI (loop-back
+ * write) and the result URI (the loop's input table, read at consume time)
+ * from it.
*
* Derived from the final (resource-allocated) schedule, so the URIs are
* exactly the ones `AssignPort` later ships to the Loop Start's input
* readers. Kept a `def`: `schedule` is a `var` that is only populated after
* `StartWorkflow`, and the first use is inside `coordinateRegionExecutors`.
*/
- private def loopStartStateUris: Map[String, String] =
+ private def loopStartPortUris: Map[String, String] =
schedule.levelSets.values.flatten.flatMap { region =>
region.getOperators.filter(_.isLoopStart).map { op =>
require(
@@ -86,7 +91,7 @@ class WorkflowExecutionManager(
s"Loop Start input port $gpid expected exactly one reader URI, " +
s"got ${cfg.storagePairs.size}"
)
- op.id.logicalOpId.id ->
VFSURIFactory.stateURI(cfg.storagePairs.head._1).toString
+ op.id.logicalOpId.id -> cfg.storagePairs.head._1.toString
}
}.toMap
@@ -132,7 +137,7 @@ class WorkflowExecutionManager(
}
executedRegions.append(nextRegions)
- val loopUris = loopStartStateUris
+ val loopUris = loopStartPortUris
Future
.collect(
nextRegions
@@ -151,7 +156,7 @@ class WorkflowExecutionManager(
coordinatorConfig,
actorService,
actorRefService,
- loopStartStateUris = loopUris
+ loopStartPortUris = loopUris
)
regionExecutionManagers(region.id)
})
diff --git
a/amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala
b/amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala
index 9da249f996..0a567b678b 100644
---
a/amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala
+++
b/amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala
@@ -253,7 +253,7 @@ class LoopIntegrationSpec
//
// This is the case that exercises the loop_counter increment/decrement and
// the loop_start_id routing on the StateFrame envelope (write addresses:
- // see InitializeExecutorRequest.loopStartStateUris): the outer loop's
state
+ // see InitializeExecutorRequest.loopStartPortUris): the outer loop's state
// passes THROUGH the inner LoopStart (+1) and inner LoopEnd (-1)
untouched,
// and is consumed only at the outer LoopEnd (counter == 0). A routing or
// counter bug would change the 4, or mis-consume and hang.
@@ -298,7 +298,7 @@ class LoopIntegrationSpec
// the LoopEnd. Before the envelope was carried through on the Scala side
// (StateFrame fields + reader/emit/save plumbing), this hop zeroed the
// counter and blanked the id, so the LoopEnd captured loop_start_id = ""
- // and the back-jump failed with "no loop-back state URI configured for
+ // and the back-jump failed with "no loop bookkeeping URI configured for
// LoopStart ''" -- the loop never iterated. Limit(10) passes every row
// through, so the loop semantics are identical to the single-loop test.
val src = textInput("1\n2\n3")
@@ -373,10 +373,10 @@ class LoopIntegrationSpec
// on both engine sides), which reaches the LoopEnd with the "no loop"
// envelope (counter 0, no LoopStart stamp) AFTER the forwarded loop
// state. The LoopEnd must pass that unstamped state through instead of
- // consuming it: consuming would clobber the captured back-jump id with ""
- // ("no loop-back state URI configured for LoopStart ''") and hand
- // run_update a State with no `table` payload (KeyError). Regression test
- // for #discussion_r3648708075 on #6661.
+ // consuming it: taking it would clobber the captured back-jump id with ""
+ // and then fail the deferred consume's bookkeeping-URI lookup ("no loop
+ // bookkeeping URI configured for LoopStart ''"). Regression test for
+ // #discussion_r3648708075 on #6661.
val src = textInput("1\n2\n3")
val start = loopStart("i = 0", "table.iloc[i]")
val mid = statefulPythonUDF()
diff --git
a/amber/src/test/python/core/architecture/handlers/control/test_initialize_executor_handler.py
b/amber/src/test/python/core/architecture/handlers/control/test_initialize_executor_handler.py
index 7b4cd2ec68..036bd3735f 100644
---
a/amber/src/test/python/core/architecture/handlers/control/test_initialize_executor_handler.py
+++
b/amber/src/test/python/core/architecture/handlers/control/test_initialize_executor_handler.py
@@ -43,10 +43,10 @@ def make_request(**kwargs) -> InitializeExecutorRequest:
def make_handler() -> InitializeExecutorHandler:
"""Wire a handler with a SimpleNamespace context exposing the fields the
- handler writes: executor_manager and loop_start_state_uris."""
+ handler writes: executor_manager and loop_start_port_uris."""
context = SimpleNamespace(
executor_manager=MagicMock(),
- loop_start_state_uris={},
+ loop_start_port_uris={},
)
return InitializeExecutorHandler(context)
@@ -60,7 +60,7 @@ class TestInitializeExecutorHandler:
"# code", False, "python"
)
- def test_stores_loop_start_state_uris_on_context(self):
+ def test_stores_loop_start_port_uris_on_context(self):
# The loop-back write addresses (LoopStart op id -> its input port's
# state URI) are per-operator setup config delivered on this RPC; the
# handler must expose them on the context for a Loop End's
@@ -68,18 +68,16 @@ class TestInitializeExecutorHandler:
handler = make_handler()
asyncio.run(
handler.initialize_executor(
- make_request(loop_start_state_uris={"loop-start-1":
"vfs:///x/state"})
+ make_request(loop_start_port_uris={"loop-start-1": "vfs:///x"})
)
)
- assert handler.context.loop_start_state_uris == {
- "loop-start-1": "vfs:///x/state"
- }
+ assert handler.context.loop_start_port_uris == {"loop-start-1":
"vfs:///x"}
def test_defaults_to_empty_map_for_plans_without_loops(self):
# betterproto defaults an absent map field to {}; the handler must
# store that default rather than leaving stale config behind (a
# recreated worker re-runs initialize_executor every region run).
handler = make_handler()
- handler.context.loop_start_state_uris = {"stale": "vfs:///old"}
+ handler.context.loop_start_port_uris = {"stale": "vfs:///old"}
asyncio.run(handler.initialize_executor(make_request()))
- assert handler.context.loop_start_state_uris == {}
+ assert handler.context.loop_start_port_uris == {}
diff --git a/amber/src/test/python/core/models/test_loop_operators.py
b/amber/src/test/python/core/models/test_loop_operators.py
index f9dad97327..6b496a5f17 100644
--- a/amber/src/test/python/core/models/test_loop_operators.py
+++ b/amber/src/test/python/core/models/test_loop_operators.py
@@ -29,10 +29,11 @@ Coverage:
- LoopEnd's process_table identity yield; condition is abstract.
- The guarded eval/exec helpers (eval_output / run_update / eval_condition)
keep the reserved `table` name out of the persistent loop state, so user
- code cannot silently clobber loop machinery. The table crosses the loop
- boundary as Arrow IPC bytes (see table_to_ipc_bytes in core.models.table);
- a user loop variable named `table` is a raised collision, not a silent
- drop (TestReservedNameCollision).
+ code cannot silently clobber loop machinery. The table never rides the
+ State content: the runtime reads it from the Loop Start's input-port
+ materialization and injects it via attach_loop_table; a user loop variable
+ named `table` is a raised collision, not a silent drop
+ (TestReservedNameCollision).
- A multi-iteration loop driven to completion through the operators and the
State to_tuple/from_tuple round-trip (TestLoopRunsToCompletion).
- The exact generated-code shape -- base64 + decode_python_template + exec,
@@ -51,7 +52,6 @@ so their handling is covered in
test_main_loop.py::TestMainLoop.
import base64
from typing import Iterator, Optional
-import pyarrow as pa
import pytest
from core.models import State, Table, TableLike, Tuple
@@ -60,7 +60,6 @@ from core.models.operator import (
LoopEndOperator,
LoopStartOperator,
)
-from core.models.table import table_from_ipc_bytes, table_to_ipc_bytes
# ---------------------------------------------------------------------------
@@ -121,9 +120,11 @@ class _StubLoopEnd(LoopEndOperator):
# ---------------------------------------------------------------------------
-def _ipc_one_row():
- """One-row table as Arrow IPC bytes (the loop `table` payload)."""
- return table_to_ipc_bytes(Table([Tuple({"v": 1})]))
+def _one_row_table() -> Table:
+ """One-row Table, standing in for the loop's input table that the runtime
+ reads from the Loop Start's input-port materialization and injects via
+ ``attach_loop_table``."""
+ return Table([Tuple({"v": 1})])
class TestLoopStartProcessState:
@@ -150,58 +151,22 @@ class TestLoopStartProcessState:
# ---------------------------------------------------------------------------
-class TestBufferedTableAccessor:
- """`TableOperator._buffered_table(port)` replaces the name-mangled
- `self._TableOperator__table_data[port]` read, so a rename of the parent
- class doesn't silently break LoopStart's table access."""
-
- def test_returns_buffered_tuples_as_table(self):
- op = _StubLoopStart()
- op.open()
- list(op.process_tuple(Tuple({"v": 1}), port=0))
- list(op.process_tuple(Tuple({"v": 2}), port=0))
-
- table = op._buffered_table(port=0)
-
- assert isinstance(table, Table)
- assert list(table.as_tuples()) == [Tuple({"v": 1}), Tuple({"v": 2})]
-
- def test_buffers_are_keyed_by_port(self):
- op = _StubLoopStart()
- op.open()
- list(op.process_tuple(Tuple({"v": 1}), port=0))
- list(op.process_tuple(Tuple({"v": 99}), port=1))
-
- assert list(op._buffered_table(port=0).as_tuples()) == [Tuple({"v":
1})]
- assert list(op._buffered_table(port=1).as_tuples()) == [Tuple({"v":
99})]
-
-
class TestLoopStartProduceStateOnFinish:
- def test_serializes_buffered_table_as_arrow_into_state_table_field(self):
- # produce_state_on_finish serializes the buffered table as an Apache
- # Arrow IPC stream, not pickle (see table_to_ipc_bytes in
- # core.models.table for why). The bytes must round-trip back to the
- # same tuples and parse as a real Arrow stream.
+ def test_produced_state_carries_only_user_variables_never_the_table(self):
+ # The input table does NOT ride the state: the LoopEnd runtime reads
+ # it from the Loop Start's input-port materialization at consume time
+ # (loopStartPortUris) and injects it via attach_loop_table. The
+ # produced state must therefore stay small, pure-JSON user variables.
op = _StubLoopStart()
op.open()
- # Drive a couple of tuples through to populate the per-port buffer.
list(op.process_tuple(Tuple({"v": 1}), port=0))
list(op.process_tuple(Tuple({"v": 2}), port=0))
produced = op.produce_state_on_finish(port=0)
assert isinstance(produced, dict)
- assert "table" in produced
- assert isinstance(produced["table"], bytes), "table must be serialized
bytes"
- # The bytes are an Arrow IPC stream (stronger than a no-pickle-prefix
- # check): if a future change swaps the encoder back to pickle, the
- # Arrow reader raises here.
- with pa.ipc.open_stream(pa.py_buffer(produced["table"])) as reader:
- reader.read_all()
- # Round-trip through the public helper must give back our two tuples.
- decoded = table_from_ipc_bytes(produced["table"])
- assert isinstance(decoded, Table)
- assert list(decoded.as_tuples()) == [Tuple({"v": 1}), Tuple({"v": 2})]
+ assert "table" not in produced
+ produced.to_tuple(0) # pure-JSON user vars: must serialize cleanly
def test_user_state_fields_survive_into_produced_state(self):
# Any vars the user set in open() (e.g. i, accumulators) must
@@ -216,8 +181,10 @@ class TestLoopStartProduceStateOnFinish:
assert produced["i"] == 0
assert produced["acc"] == []
# loop_counter is no longer seeded into the operator's state; it is
- # runtime-owned and rides on the StateFrame envelope.
+ # runtime-owned and rides on the StateFrame envelope. The table is
+ # injected at the LoopEnd by the runtime, never embedded here.
assert "loop_counter" not in produced
+ assert "table" not in produced
# ---------------------------------------------------------------------------
@@ -261,6 +228,20 @@ class TestLoopEndBase:
# None is what condition() short-circuits on.
assert op._loop_table is None
assert op.condition() is False
+ # Attaching the table alone (what the runtime does right before the
+ # consume) must NOT mark the loop as consumed: only a successful
+ # run_update does.
+ op.attach_loop_table(_one_row_table())
+ assert op._loop_table is None
+ assert op.condition() is False
+
+ def test_run_update_fails_loud_when_no_table_attached(self):
+ # The runtime must attach the loop's input table before the matching
+ # consume; a missing attach is a runtime wiring bug and must not
+ # surface as a confusing NameError from the user's expression.
+ op = _StubLoopEnd(update="i += 1")
+ with pytest.raises(RuntimeError, match="not attached"):
+ op.process_state(State({"i": 0}), port=0)
# ---------------------------------------------------------------------------
@@ -275,17 +256,13 @@ class TestLoopEndMatchingBranch:
# state flows downstream; the actual loop-back is driven by
# main_loop.complete() reading executor.state.
op = _StubLoopEnd(update="i += 1", condition_expr="i < 3")
- # Simulate LoopStart's produced state arriving here. The table rides as
- # Arrow IPC bytes (see produce_state_on_finish), not pickle.
- # The content carries only user data (i) and the per-iteration table
- # scratch. loop_counter / LoopStartId are runtime-owned and ride the
- # StateFrame envelope, never the content.
- incoming = State(
- {
- "i": 1,
- "table": _ipc_one_row(),
- }
- )
+ # Simulate the runtime's consume: it reads the loop's input table from
+ # the Loop Start's input-port materialization and attaches it, then
+ # hands over LoopStart's produced state (pure user variables --
+ # loop_counter / LoopStartId are runtime-owned and ride the StateFrame
+ # envelope, never the content).
+ op.attach_loop_table(_one_row_table())
+ incoming = State({"i": 1})
result = op.process_state(incoming, port=0)
@@ -299,17 +276,16 @@ class TestLoopEndMatchingBranch:
assert op.condition() is True # i became 2, 2 < 3
# Run another iteration to push i past the threshold.
- op.process_state(
- State(
- {
- "i": 2,
- "table": _ipc_one_row(),
- }
- ),
- port=0,
- )
+ op.attach_loop_table(_one_row_table())
+ op.process_state(State({"i": 2}), port=0)
assert op.condition() is False # i became 3, 3 < 3 is False
+ # Each update TAKES its attach: a further update without a fresh
+ # attach must fail loud rather than silently reuse the previous
+ # table (the clear in run_update is what this pins).
+ with pytest.raises(RuntimeError, match="not attached"):
+ op.process_state(State({"i": 3}), port=0)
+
# ---------------------------------------------------------------------------
# Nested-loop counter behaviour -- LoopStart +1, LoopEnd -1, and the
@@ -370,6 +346,10 @@ class TestLoopRunsToCompletion:
update="total += int(table.iloc[i]['v']); i += 1; output =
total",
condition_expr="i < len(table)",
)
+ # The runtime reads the loop's input table from the Loop Start's
+ # input-port materialization (the same rows the LoopStart
+ # buffered) and attaches it before the consume.
+ end.attach_loop_table(Table(rows))
end.process_state(forwarded, port=0)
if not end.condition():
break
@@ -419,7 +399,8 @@ class TestReservedNameCollision:
def test_loop_end_raises_when_update_rebinds_table(self):
# `update` rebinds `table`, which run_update would otherwise strip.
op = _StubLoopEnd(update="table = 1")
- incoming = State({"i": 1, "table": _ipc_one_row()})
+ op.attach_loop_table(_one_row_table())
+ incoming = State({"i": 1})
with pytest.raises(ValueError, match="'table' is reserved by the loop
runtime"):
op.process_state(incoming, port=0)
@@ -428,7 +409,8 @@ class TestReservedNameCollision:
# still flag the reserved-name collision (namespace.get) rather than
# escape as a bare KeyError on the missing key.
op = _StubLoopEnd(update="del table")
- incoming = State({"i": 1, "table": _ipc_one_row()})
+ op.attach_loop_table(_one_row_table())
+ incoming = State({"i": 1})
with pytest.raises(ValueError, match="'table' is reserved by the loop
runtime"):
op.process_state(incoming, port=0)
@@ -457,15 +439,15 @@ class TestLoopExpressionScoping:
# `update` assigns from a genexp whose body references the loop
# variable `base`.
op = _StubLoopEnd(update="total = sum(v + base for v in [1, 2, 3])")
- incoming = State({"base": 10, "table": _ipc_one_row()})
- op.process_state(incoming, port=0)
+ op.attach_loop_table(_one_row_table())
+ op.process_state(State({"base": 10}), port=0)
assert op.state["total"] == 36
def test_run_update_resolves_lambda_capturing_loop_vars(self):
# A lambda in `update` closes over the loop variable `offset`.
op = _StubLoopEnd(update="ranked = sorted([3, 1, 2], key=lambda e: e -
offset)")
- incoming = State({"offset": 0, "table": _ipc_one_row()})
- op.process_state(incoming, port=0)
+ op.attach_loop_table(_one_row_table())
+ op.process_state(State({"offset": 0}), port=0)
assert op.state["ranked"] == [1, 2, 3]
def test_eval_condition_resolves_genexp_over_loop_vars(self):
@@ -474,8 +456,8 @@ class TestLoopExpressionScoping:
op = _StubLoopEnd(
update="i += 1", condition_expr="all(x > floor for x in [1, 2, 3])"
)
- incoming = State({"i": 0, "floor": 0, "table": _ipc_one_row()})
- op.process_state(incoming, port=0)
+ op.attach_loop_table(_one_row_table())
+ op.process_state(State({"i": 0, "floor": 0}), port=0)
assert op.condition() is True
def test_run_initialization_resolves_genexp_over_init_vars(self):
@@ -501,8 +483,8 @@ class TestLoopExpressionScoping:
def test_updated_state_has_no_builtins_leak(self):
op = _StubLoopEnd(update="i += 1")
- incoming = State({"i": 0, "table": _ipc_one_row()})
- op.process_state(incoming, port=0)
+ op.attach_loop_table(_one_row_table())
+ op.process_state(State({"i": 0}), port=0)
assert "__builtins__" not in op.state
op.state.to_tuple(0) # must not raise
@@ -584,7 +566,8 @@ class TestGeneratedCodeShape:
exec(source, namespace)
op = namespace["ProcessLoopEndOperator"]()
- incoming = State({"i": 1, "note": "it's", "table": _ipc_one_row()})
+ op.attach_loop_table(_one_row_table())
+ incoming = State({"i": 1, "note": "it's"})
assert op.process_state(incoming, port=0) is None
assert op.state["i"] == 2 # update ran
assert op.condition() is True # quoted condition round-tripped
diff --git a/amber/src/test/python/core/runnables/test_main_loop.py
b/amber/src/test/python/core/runnables/test_main_loop.py
index 1da2d19691..5aa259c428 100644
--- a/amber/src/test/python/core/runnables/test_main_loop.py
+++ b/amber/src/test/python/core/runnables/test_main_loop.py
@@ -28,8 +28,10 @@ from threading import Thread
from core.models import (
DataFrame,
InternalQueue,
+ Schema,
State,
StateFrame,
+ Table,
Tuple,
)
from core.models.internal_queue import (
@@ -38,6 +40,7 @@ from core.models.internal_queue import (
ECMElement,
)
from core.models.operator import LoopEndOperator, LoopStartOperator
+from core.storage.vfs_uri_factory import VFSURIFactory
from core.runnables import MainLoop
from core.util import set_one_of
from proto.org.apache.texera.amber.core import (
@@ -2211,7 +2214,7 @@ class TestMainLoop:
executor.state = State({"i": 1})
main_loop.context.executor_manager.executor = executor
main_loop._loop_start_id = "loop-start-1"
- main_loop.context.loop_start_state_uris = {"loop-start-1":
"vfs:///x/state"}
+ main_loop.context.loop_start_port_uris = {"loop-start-1": "vfs:///x"}
console_msgs = []
pauses = []
@@ -2486,18 +2489,23 @@ class TestMainLoop:
# the outer loop's id rides through unchanged
assert emitted_id == "outer-loop"
- def test_loopend_consume_invokes_operator_at_counter_zero(
+ def test_loopend_consume_defers_operator_to_end_channel(
self, main_loop, monkeypatch
):
- # loop_counter == 0 is the matching loop: the runtime runs the operator
- # (consume) via the context switch. The operator returns None, so no
- # state is emitted; the loop-back is driven by complete() separately.
+ # loop_counter == 0 is the matching loop. The runtime STASHES the state
+ # here and runs the operator at EndChannel instead
+ # (_consume_pending_loop_state): the loop's input table is read from
the
+ # Loop Start's input-port materialization, and that read must not
+ # overlap this worker's own materialization reader, which is still
+ # streaming at consume time. Nothing observable moves: the matching
+ # consume emits no state downstream either way.
# Reviewer feedback (#discussion_r3285892237): the envelope's loop
# metadata (loop_counter / loop_start_id) is internal runtime data --
# the runtime captures it onto its own instance state, and the
# user-facing State handed to the operator carries only the inner
# State's keys, never the envelope names.
- main_loop.context.executor_manager.executor = _FalseLoopEnd()
+ executor = _FalseLoopEnd()
+ main_loop.context.executor_manager.executor = executor
emitted, switched, reset_calls = self._capture_state_emit(
main_loop, monkeypatch
)
@@ -2507,17 +2515,29 @@ class TestMainLoop:
"get_output_state",
lambda: None,
)
+ # Stub the runtime's table read (the real one opens the Loop Start's
+ # input-port materialization; pinned by the jump/read URI tests).
+ loop_table = Table([Tuple({"v": 1})])
+ reads = []
+
+ def _read():
+ reads.append(True)
+ return loop_table
+
+ monkeypatch.setattr(main_loop, "_read_loop_input_table", _read)
+ incoming = State({"i": 42, "acc": [1, 2, 3]})
main_loop._process_state_frame(
- StateFrame(
- State({"i": 42, "acc": [1, 2, 3]}),
- loop_counter=0,
- loop_start_id="outer-loop",
- )
+ StateFrame(incoming, loop_counter=0, loop_start_id="outer-loop")
)
- assert switched == [True], "consume branch must invoke the operator"
- assert emitted == [], "operator returned None -> nothing emitted"
+ # At consume: state stashed, operator NOT invoked, table NOT read yet
+ # (no storage I/O while this worker's reader is still streaming).
+ assert switched == [], "consume must not invoke the operator yet"
+ assert reads == [], "the table must not be read at consume time"
+ assert main_loop._pending_loop_state is incoming
+ assert executor._attached_table is None
+ assert emitted == [], "the matching consume emits no state downstream"
assert reset_calls == [], "consume / single loop must not reset output"
# The runtime captured the envelope metadata onto its own instance
# state...
@@ -2525,13 +2545,25 @@ class TestMainLoop:
# ...but never wrote it into the user-facing State the operator sees.
# (The consume branch sets `current_input_state` BEFORE the stubbed
# context switch, so this is exactly what the operator would receive.)
- passed_to_operator = (
- main_loop.context.state_processing_manager.current_input_state
+ # ...and the state it stashed for the operator carries only the inner
+ # State's keys, never the envelope names.
+ assert set(main_loop._pending_loop_state.keys()) == {"i", "acc"}
+ assert "loop_start_id" not in main_loop._pending_loop_state
+ assert "loop_counter" not in main_loop._pending_loop_state
+
+ # Then at EndChannel the deferred consume runs: the table is read once
+ # (the reader has finished by now) and handed to the operator.
+ consumed = []
+ monkeypatch.setattr(
+ executor, "process_state", lambda st, port: consumed.append((st,
port))
)
- assert isinstance(passed_to_operator, State)
- assert set(passed_to_operator.keys()) == {"i", "acc"}
- assert "loop_start_id" not in passed_to_operator
- assert "loop_counter" not in passed_to_operator
+
+ main_loop._consume_pending_loop_state(executor)
+
+ assert reads == [True], "the table is read exactly once, at EndChannel"
+ assert executor._attached_table is loop_table
+ assert consumed == [(incoming, 0)]
+ assert main_loop._pending_loop_state is None, "stash must be cleared"
def test_loopend_forwards_unstamped_state_without_consuming(
self, main_loop, monkeypatch
@@ -2539,9 +2571,10 @@ class TestMainLoop:
# A loop-body operator that emits its own boundary state
# (produce_state_on_start/finish -- a public API on both engine sides)
# sends it with the "no loop" envelope (counter 0, id ""). That state
- # is NOT the loop's own boundary state: consuming it would clobber the
- # captured back-jump id with "" and hand run_update a State with no
- # `table` payload (KeyError). A real loop state is always stamped --
+ # is NOT the loop's own boundary state: taking it would clobber the
+ # captured back-jump id with "" and then fail the deferred consume's
+ # bookkeeping-URI lookup for LoopStart ''. A real loop state is always
+ # stamped --
# the matching LoopStart stamps its own id on every iteration's output
# -- so an UNstamped counter-0 frame at a LoopEnd must be forwarded
# downstream unchanged, skipping the operator, like any default
@@ -2680,10 +2713,12 @@ class TestMainLoop:
# A loop body may branch and converge on the Loop End, so its input
# port takes fan-in. Every reader on that port replays its own
# branch's states, so the SAME iteration's state arrives once per
- # branch. Only the first may be consumed: running the user's `update`
- # again would advance the loop variables once per branch (e.g. `i += 1`
- # twice), ending the loop early with wrong results.
- main_loop.context.executor_manager.executor = _FalseLoopEnd()
+ # branch. The stash must take the FIRST and drop the rest -- a later
+ # arrival silently overwriting _pending_loop_state would make the
+ # consumed copy depend on reader scheduling. Assert through the whole
+ # deferred path, not just the stash.
+ executor = _FalseLoopEnd()
+ main_loop.context.executor_manager.executor = executor
emitted, switched, reset_calls = self._capture_state_emit(
main_loop, monkeypatch
)
@@ -2692,24 +2727,39 @@ class TestMainLoop:
"get_output_state",
lambda: None,
)
+ monkeypatch.setattr(
+ main_loop, "_read_loop_input_table", lambda: Table([Tuple({"v":
1})])
+ )
+ consumed = []
+ monkeypatch.setattr(
+ executor, "process_state", lambda st, port: consumed.append((st,
port))
+ )
- def deliver():
+ first = State({"i": 42})
+ second = State({"i": 42})
+
+ def deliver(state):
main_loop._process_state_frame(
- StateFrame(
- State({"i": 42}),
- loop_counter=0,
- loop_start_id="outer-loop",
- )
+ StateFrame(state, loop_counter=0, loop_start_id="outer-loop")
)
- deliver() # branch A
- deliver() # branch B replays the same iteration's state
+ deliver(first) # branch A
+ deliver(second) # branch B replays the same iteration's state
- assert switched == [True], "the operator must consume exactly once"
+ assert main_loop._loop_state_consumed is True
+ assert main_loop._pending_loop_state is first, "the duplicate must not
stash"
assert emitted == [], "a consume emits nothing downstream, duplicate
or not"
assert reset_calls == []
assert main_loop._loop_start_id == "outer-loop"
- assert main_loop._loop_state_consumed is True
+
+ main_loop._consume_pending_loop_state(executor)
+
+ assert switched == [], "the deferred consume does not switch context"
+ assert consumed == [(first, 0)], "the operator must update exactly
once"
+ # The consume re-arms the duplicate guard (everything on the port is
+ # already in by EndChannel), making the flag per-execution by
+ # construction rather than by worker recreation.
+ assert main_loop._loop_state_consumed is False
# ------------------------------------------------------------------ #
# _jump_to_loop_start
@@ -2719,7 +2769,7 @@ class TestMainLoop:
# stamps is now computed inline in process_input_state via the
# canonical `get_logical_op_id` (pinned by that helper's own suite),
# and the loop-back write address is not computed worker-side at all:
- # it is setup config (InitializeExecutorRequest.loopStartStateUris --
+ # it is setup config (InitializeExecutorRequest.loopStartPortUris --
# see the proto comment for the full story).
# ------------------------------------------------------------------ #
@@ -2769,16 +2819,220 @@ class TestMainLoop:
_create,
)
+ @pytest.mark.timeout(2)
+ def test_complete_flushes_prints_from_the_user_condition(
+ self, main_loop, monkeypatch
+ ):
+ # complete() flushes console messages on entry, before condition()
+ # runs, and then shuts the worker down -- so without a second flush a
+ # print() inside the user's condition would be captured and never sent.
+ #
+ # Mirrors LoopEndOpDesc.generatePythonCode: the user's text goes
+ # through eval_condition, i.e. eval() against a bare namespace dict --
+ # NOT a method of a module-level class -- so the print executes in a
+ # frame whose globals have no __name__. The capture must survive that
+ # (replace_print looks the module name up with .get); a plain method
+ # here would pass even with a capture that crashes on the generated
+ # path.
+ class _PrintingLoopEnd(LoopEndOperator):
+ def condition(self):
+ return self.eval_condition("print('hello from condition') or
False")
+
+ executor = _PrintingLoopEnd()
+ # eval_condition short-circuits to False before a consume; run a
+ # minimal successful update first so the user expression actually
+ # evaluates (this is the state a real matching consume leaves behind).
+ executor.attach_loop_table(Table([Tuple({"v": 1})]))
+ executor.run_update("pass", State())
+ main_loop.context.executor_manager.executor = executor
+
+ console_msgs = []
+ monkeypatch.setattr(
+ main_loop, "_send_console_message", lambda msg:
console_msgs.append(msg)
+ )
+ monkeypatch.setattr(main_loop.data_processor, "stop", lambda: None)
+ monkeypatch.setattr(
+ main_loop.context.state_manager, "transit_to", lambda state: None
+ )
+ monkeypatch.setattr(main_loop.context, "close", lambda: None)
+
+ class _Coordinator:
+ def worker_execution_completed(self, request):
+ pass
+
+ monkeypatch.setattr(
+ main_loop._async_rpc_client, "coordinator_stub", lambda:
_Coordinator()
+ )
+
+ main_loop.complete()
+
+ printed = [m for m in console_msgs if "hello from condition" in
m.title]
+ assert printed, (
+ "a print() in the user's condition must reach the console before "
+ f"the worker completes; sent: {[m.title for m in console_msgs]}"
+ )
+
+ def test_deferred_consume_captures_user_prints(self, main_loop,
monkeypatch):
+ # The `update` runs on the main loop thread, outside
+ # DataProcessor._executor_session, so its print capture has to be
+ # applied explicitly here -- otherwise a print() in the user's update
+ # goes to the worker's stdout and never reaches the console.
+ #
+ # Mirrors LoopEndOpDesc.generatePythonCode: the user's text goes
+ # through run_update, i.e. exec() against a bare namespace dict, so
+ # the print executes in a frame whose globals have no __name__ and the
+ # capture must survive that (replace_print looks the module name up
+ # with .get). A plain print() in an overridden method here would pass
+ # even with a capture that crashes on the generated path.
+ class _PrintingLoopEnd(LoopEndOperator):
+ def condition(self):
+ return self.eval_condition("False")
+
+ def process_state(self, state, port):
+ self.run_update("print('hello from update')\ni += 1", state)
+ return None
+
+ executor = _PrintingLoopEnd()
+ main_loop.context.executor_manager.executor = executor
+ main_loop._pending_loop_state = State({"i": 1})
+ monkeypatch.setattr(
+ main_loop, "_read_loop_input_table", lambda: Table([Tuple({"v":
1})])
+ )
+
+ main_loop._consume_pending_loop_state(executor)
+
+ printed = [
+ msg
+ for msg in main_loop.context.console_message_manager.get_messages(
+ force_flush=True
+ )
+ if "hello from update" in msg.title
+ ]
+ assert printed, "the user's print must be captured as a console
message"
+ assert printed[0].msg_type == ConsoleMessageType.PRINT
+
+ def test_read_loop_input_table_opens_the_result_uri_of_the_configured_base(
+ self, main_loop, monkeypatch
+ ):
+ # The other half of the base-URI split (the jump test pins the state
+ # URI): the loop's input table is read from result_uri(base) of the
+ # SAME configured base, through DocumentFactory.open_document.
+ main_loop._loop_start_id = "outer-loop"
+ main_loop.context.loop_start_port_uris = {"outer-loop":
"vfs:///wf/port/outer"}
+
+ opened = []
+ rows = [Tuple({"v": 1}), Tuple({"v": 2})]
+ schema = Schema(raw_schema={"v": "LONG"})
+ casts = []
+ for tup in rows:
+ monkeypatch.setattr(
+ tup,
+ "cast_to_schema",
+ lambda s, _t=tup: casts.append((_t, s)),
+ raising=True,
+ )
+
+ class _Doc:
+ def get(self):
+ return iter(rows)
+
+ monkeypatch.setattr(
+ "core.runnables.main_loop.DocumentFactory.open_document",
+ lambda uri: (opened.append(uri) or (_Doc(), schema)),
+ )
+
+ table = main_loop._read_loop_input_table()
+
+ assert opened == [VFSURIFactory.result_uri("vfs:///wf/port/outer")]
+ assert isinstance(table, Table)
+ assert list(table.as_tuples()) == rows
+ # Every tuple is normalized to the doc's schema, exactly like the
+ # input-port reader that streams this same doc.
+ assert casts == [(tup, schema) for tup in rows]
+
+ def test_read_loop_input_table_raises_when_uri_not_configured(
+ self, main_loop, monkeypatch
+ ):
+ # Same fail-loud contract as the back-edge write: a LoopEnd whose
+ # captured id has no setup-config entry must raise rather than read
+ # from a guessed location.
+ main_loop._loop_start_id = "outer-loop"
+ main_loop.context.loop_start_port_uris = {}
+ opened = []
+ monkeypatch.setattr(
+ "core.runnables.main_loop.DocumentFactory.open_document",
+ lambda uri: (opened.append(uri) or (None, None)),
+ )
+
+ with pytest.raises(RuntimeError, match="no loop bookkeeping URI"):
+ main_loop._read_loop_input_table()
+
+ assert opened == [], "must fail before touching storage"
+
+ @pytest.mark.timeout(2)
+ def test_end_channel_holds_the_region_when_the_deferred_consume_fails(
+ self, main_loop, monkeypatch
+ ):
+ # The deferred consume runs the table read and the user's `update`.
+ # Both can fail, and the failure must hold the region: complete() is
+ # the tail of _process_end_channel, so reporting the error there would
+ # arrive after port_completed had already been sent for every port and
+ # would read as a false success (region completion is port-based).
+ class _BoomLoopEnd(LoopEndOperator):
+ def condition(self):
+ return False
+
+ def process_state(self, state, port):
+ raise ValueError("name 'i' is not defined")
+
+ executor = _BoomLoopEnd()
+ main_loop.context.executor_manager.executor = executor
+ main_loop._pending_loop_state = State({"i": 1})
+ monkeypatch.setattr(
+ main_loop, "_read_loop_input_table", lambda: Table([Tuple({"v":
1})])
+ )
+
+ completed = []
+ port_completed_calls = []
+ console_msgs = []
+ monkeypatch.setattr(main_loop, "process_input_state", lambda *a, **k:
None)
+ monkeypatch.setattr(main_loop, "process_input_tuple", lambda: None)
+ monkeypatch.setattr(main_loop, "complete", lambda:
completed.append(True))
+ monkeypatch.setattr(
+ main_loop, "_send_console_message", lambda msg:
console_msgs.append(msg)
+ )
+ monkeypatch.setattr(
+ main_loop.context.pause_manager,
+ "pause",
+ lambda pause_type, change_state=True: None,
+ )
+
+ class _Coordinator:
+ def port_completed(self, request):
+ port_completed_calls.append(request)
+
+ monkeypatch.setattr(
+ main_loop._async_rpc_client, "coordinator_stub", lambda:
_Coordinator()
+ )
+
+ main_loop._process_end_channel()
+
+ assert main_loop.context.exception_manager.has_exception()
+ error_msgs = [m for m in console_msgs if m.msg_type ==
ConsoleMessageType.ERROR]
+ assert len(error_msgs) == 1
+ assert "name 'i' is not defined" in error_msgs[0].title
+ assert port_completed_calls == [], "no port may be reported complete"
+ assert completed == [], "the worker must not complete"
+
def test_jump_to_loop_start_sends_rpc_then_writes_state_in_order(
self, main_loop, monkeypatch
):
# One shared event log for the jump RPC and the storage calls, so
# the cross-channel ordering is pinned along with each contract.
main_loop._loop_start_id = "outer-loop"
- # The write address is setup-injected config keyed by the captured id.
- main_loop.context.loop_start_state_uris = {
- "outer-loop": "vfs:///wf/state/outer"
- }
+ # The bookkeeping BASE URI is setup-injected config keyed by the
+ # captured id; the write address is derived from it (state_uri).
+ main_loop.context.loop_start_port_uris = {"outer-loop":
"vfs:///wf/port/outer"}
events = []
self._patch_create_document(monkeypatch, events)
@@ -2800,14 +3054,15 @@ class TestMainLoop:
assert kind == "jump"
assert request.target_operator_id.id == "outer-loop"
# (ii) Then the exact iceberg write contract, in order:
- # create_document with the configured URI and State.SCHEMA, open
+ # create_document with the state URI DERIVED from the configured base
+ # URI and State.SCHEMA, open
# writer("0"), a single put_one with the State as a depth-0 tuple
# (the back-edge fires only after the matching LoopEnd consumed at
# loop_counter == 0, so the next iteration starts at depth 0),
# then close. The tuple object's internals are exercised elsewhere.
assert events[1] == (
"create_document",
- "vfs:///wf/state/outer",
+ VFSURIFactory.state_uri("vfs:///wf/port/outer"),
State.SCHEMA,
)
assert events[2] == ("writer", "0")
@@ -2824,7 +3079,7 @@ class TestMainLoop:
# RPC and before any storage write -- rewinding the schedule
# without a back-edge write would hang the loop.
main_loop._loop_start_id = "outer"
- main_loop.context.loop_start_state_uris = {}
+ main_loop.context.loop_start_port_uris = {}
rpc_calls = []
write_log = []
@@ -2833,7 +3088,7 @@ class TestMainLoop:
class _Executor:
state = State({"i": 7})
- with pytest.raises(RuntimeError, match="no loop-back state URI"):
+ with pytest.raises(RuntimeError, match="no loop bookkeeping URI"):
main_loop._jump_to_loop_start(
_Executor(), self._stub_coordinator(rpc_calls)
)
diff --git
a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/WorkerSpec.scala
b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/WorkerSpec.scala
index 7093dbfe8e..11cf67ab87 100644
---
a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/WorkerSpec.scala
+++
b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/WorkerSpec.scala
@@ -201,7 +201,7 @@ class WorkerSpec
"org.apache.texera.amber.engine.architecture.worker.DummyOperatorExecutor"
),
isSource = false,
- loopStartStateUris = Map.empty
+ loopStartPortUris = Map.empty
),
AsyncRPCContext(COORDINATOR, identifier1),
4
diff --git
a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/WorkflowWorkerSpec.scala
b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/WorkflowWorkerSpec.scala
index a5b31b4801..277ff486cd 100644
---
a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/WorkflowWorkerSpec.scala
+++
b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/WorkflowWorkerSpec.scala
@@ -119,7 +119,7 @@ class WorkflowWorkerSpec
"org.apache.texera.amber.engine.architecture.worker.DummyOperatorExecutor"
),
isSource = false,
- loopStartStateUris = Map.empty
+ loopStartPortUris = Map.empty
)
)
diff --git
a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/managers/SerializationManagerSpec.scala
b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/managers/SerializationManagerSpec.scala
index 30f91ddf79..5c6278a1c0 100644
---
a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/managers/SerializationManagerSpec.scala
+++
b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/managers/SerializationManagerSpec.scala
@@ -57,7 +57,7 @@ class SerializationManagerSpec extends AnyFlatSpec {
totalWorkerCount = totalWorkers,
opExecInitInfo = info,
isSource = false,
- loopStartStateUris = Map.empty
+ loopStartPortUris = Map.empty
)
"SerializationManager.restoreExecutorState" should
diff --git
a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/PhysicalOp.scala
b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/PhysicalOp.scala
index a1e9305066..c4dc2307ee 100644
---
a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/PhysicalOp.scala
+++
b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/PhysicalOp.scala
@@ -208,8 +208,9 @@ case class PhysicalOp(
// restricting it to only the requiring operator's regions is a possible
// future optimization. Default false.
requiresMaterializedExecution: Boolean = false,
- // Marks the Loop Start operator of a loop; the scheduler resolves the
loop-back
- // write address from it (see
InitializeExecutorRequest.loopStartStateUris). Default false.
+ // Marks the Loop Start operator of a loop; the scheduler resolves the
loop's
+ // bookkeeping base URI from it (loop-back write address + input-table
read;
+ // see InitializeExecutorRequest.loopStartPortUris). Default false.
isLoopStart: Boolean = false,
// hint for number of workers
suggestedWorkerNum: Option[Int] = None,
diff --git
a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/loop/LoopEndOpDescSpec.scala
b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/loop/LoopEndOpDescSpec.scala
index c8dc04688f..2c87aeba64 100644
---
a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/loop/LoopEndOpDescSpec.scala
+++
b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/loop/LoopEndOpDescSpec.scala
@@ -96,11 +96,12 @@ class LoopEndOpDescSpec extends AnyFlatSpec with
LoopOpDescSpecMixin {
}
it should "not exec user code inline against self.state (the guard lives in
the base helpers)" in {
- // The table decode (Arrow IPC) and the user update/condition exec run in
- // the LoopEnd base helpers (run_update / eval_condition) against a
- // throwaway namespace, so the reserved `table` never persists in the loop
- // state (a user rebind raises). The generated operator must not touch it
- // directly or exec user code against self.state.
+ // The user update/condition exec runs in the LoopEnd base helpers
+ // (run_update / eval_condition) against a throwaway namespace seeded with
+ // the runtime-attached input table, so the reserved `table` never
+ // persists in the loop state (a user rebind raises). The generated
+ // operator must not touch it directly or exec user code against
+ // self.state.
val code = desc(update = "i = i + 7", condition = "i <
3").generatePythonCode()
code should not include "exec("
code should not include "self.state[\"table\"]"
diff --git
a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/loop/LoopStartOpDescSpec.scala
b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/loop/LoopStartOpDescSpec.scala
index 94451b6ed7..d6853cd9c4 100644
---
a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/loop/LoopStartOpDescSpec.scala
+++
b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/loop/LoopStartOpDescSpec.scala
@@ -118,7 +118,7 @@ class LoopStartOpDescSpec extends AnyFlatSpec with
LoopOpDescSpecMixin {
it should "mark the physical op as the loop start" in {
// The scheduler resolves each Loop Start's loop-back write address (the
// state URI of its input port) from this flag and delivers it to workers
- // at setup via InitializeExecutorRequest.loopStartStateUris.
+ // at setup via InitializeExecutorRequest.loopStartPortUris.
desc().getPhysicalOp(workflowId, executionId).isLoopStart shouldBe true
}
