Xiao-zhen-Liu commented on code in PR #6971:
URL: https://github.com/apache/texera/pull/6971#discussion_r3698118696
##########
amber/src/test/python/core/runnables/test_main_loop.py:
##########
@@ -2590,16 +2636,189 @@ def _create(uri, schema):
_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.
+ class _PrintingLoopEnd(LoopEndOperator):
Review Comment:
Both of these pass for the wrong reason, which is how the `KeyError` I
flagged on `main_loop.py:213` survives a green suite.
`_PrintingLoopEnd` here (and in
`test_deferred_consume_captures_user_prints`) defines `condition` /
`process_state` as ordinary methods on a class in this module, so the print
executes in a frame whose `f_globals` has `__name__` and capture works fine.
The shipped operator doesn't take that path: `LoopEndOpDesc.generatePythonCode`
emits `return self.eval_condition(...)` and `self.run_update(...)`, which
`eval`/`exec` the user's text against a bare dict -- and that's where it raises.
A test that drives the generated `ProcessLoopEndOperator` with a printing
expression would have caught it, and would keep catching it.
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -412,6 +493,21 @@ def _process_start_channel(self) -> None:
def _process_end_channel(self) -> None:
self.process_input_state()
+ # 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. Running it before the has_exception() hold below
+ # puts it under the same guard as a state-emission failure. This is
+ # still past the reader: EndChannel means it finished streaming.
+ executor = self.context.executor_manager.executor
Review Comment:
The consume runs before the `has_exception()` check below it, so if
`process_input_state()` already reported an error, the worker still does an
iceberg/S3 read and runs the user's `update` before bailing -- and can report a
second, more confusing error on top of the first.
Hoisting the `has_exception()` check above this block is free. Low practical
risk today (a Loop End's `produce_state_on_finish` is the base no-op, so
there's rarely anything to report at this point), but nothing pins the order
either way -- moving this block to after the hold leaves the suite green.
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -85,14 +88,19 @@ def __init__(
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
Review Comment:
The premise checks out -- I traced the worker teardown/recreate path -- so
this isn't a live bug. But it puts a loop correctness property on a scheduler
behavior several layers away in Scala that no test pins, and respawning a
Python worker per iteration is exactly the kind of thing someone profiles and
"optimizes" later. `LoopIntegrationSpec` already calls out the respawn cost.
Clearing the flag in `_consume_pending_loop_state`, next to
`self._pending_loop_state = None`, costs one line and makes it per-execution by
construction. It's safe for the same port-alignment reason the deferral is
safe: nothing can arrive on the port after the aligned EndChannel, so every
duplicate is already in by the time the consume runs.
##########
amber/src/main/python/core/models/operator.py:
##########
@@ -452,21 +443,44 @@ def __init__(self):
# 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
Review Comment:
"so the missing-table guard fires on every iteration rather than only the
first" contradicts the premise you rely on over in `main_loop.py:96` -- that
workers are recreated per region execution, so a `LoopEndOperator` instance
only ever handles one iteration. There's no second iteration in the same
instance for the guard to fire on.
The clear is still worth keeping, just for a different reason (defensive:
nothing downstream can reuse a table it wasn't handed). Worth saying that
instead. Deleting the clear also leaves the suite green --
`test_loop_operators.py:238` only covers the never-attached case; one extra
assert in the two-iteration test would cover attach -> update ->
update-without-attach.
##########
amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/rpc/controlcommands.proto:
##########
@@ -254,12 +254,18 @@ 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 -> BASE URI of
+ // that Loop Start's input port materialization. Constant per execution
Review Comment:
"BASE URI of that Loop Start's input port materialization" sends a reader
looking for a config keyed by the Loop Start's own port. It's the *upstream
output port's* base URI, which the Loop Start's input port reads from --
`cfg.storagePairs.head._1` in `WorkflowExecutionManager` is where the
difference is visible. Same wording in the Scala doc there. One clause in each
place.
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -103,21 +111,73 @@ def __init__(
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, _ = DocumentFactory.open_document(result_uri)
Review Comment:
This drops the schema `open_document` returns and the tuples go straight
into `Table(...)`. The only other reader of this exact doc --
`InputPortMaterializationReaderRunnable` -- calls
`tup.cast_to_schema(self.tuple_schema)` on every tuple before it reaches the
operator.
Harmless today, since the Arrow provider already hands back `.as_py()`
values and the coercions are no-ops. But two readers of the same document
normalizing differently is the kind of thing that bites after an unrelated
storage change. Either cast, or say in the comment why it isn't needed here.
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -382,20 +453,30 @@ def _process_state_frame(self, frame: StateFrame) -> None:
if isinstance(executor, LoopEndOperator):
# Matching LoopEnd (in_counter == 0): it will consume this state
# and jump back. Remember which LoopStart to jump to (it rides
- # the envelope) for complete()/_jump_to_loop_start.
+ # 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. Take only
+ # the first: running the user's `update` again would advance the
Review Comment:
The stated reason doesn't hold. `run_update` seeds its exec namespace from
the *incoming* state, not from `self.state` (`operator.py:481`, same on main),
so re-running it on an identical copy is idempotent -- `i += 1` gives 43 both
times. And in the deferred design the operator is invoked once regardless:
`_consume_pending_loop_state` runs once at EndChannel, so without the flag a
second copy would only overwrite `_pending_loop_state`.
The flag's real effect is first-wins instead of last-wins. That's worth
having, but say that -- someone who checks this rationale will find it doesn't
hold and may conclude the flag is dead code.
Separately, "the copies are identical" isn't enforced anywhere.
`Operator.process_state` is a plain overridable method (`operator.py:108-116`),
so a UDF in one branch can return a modified state, and then which copy wins
depends on arrival order. Either say state-modifying operators inside a loop
body aren't supported, or drop the identity claim and say you take the first
deliberately.
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -148,12 +208,23 @@ def complete(self) -> None:
# 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(
Review Comment:
This is the one I'd fix before merge. Wrapping `condition()` in
`replace_print` makes a `print()` inside a Loop End condition raise `KeyError:
'__name__'` and stop the loop.
`replace_print` builds its source string from
`inspect.currentframe().f_back.f_globals['__name__']` (`replace_print.py:75`),
and the condition runs as `eval(expr, namespace)` where `namespace` is a bare
dict (`operator.py:340-341`). `eval`/`exec` inject `__builtins__` but never
`__name__`. Reproduced directly:
```
eval("print('dbg') or i < 3", {"i": 1}) -> KeyError: '__name__'
exec("i += 1\nprint('upd')", {"i": 1}) -> KeyError: '__name__'
```
The bug itself isn't yours -- `replace_print.py` has no changed lines in
this PR, and on main the same crash already hits the Loop End `update` and the
Loop Start `output` expression, both of which ran inside `_executor_session`'s
`replace_print`. What this PR changes is the `condition` field: on main a print
there went to worker stdout and was lost; now it raises, `complete()` reports
it, and the back-edge never fires, so the loop ends after one iteration with a
`KeyError` the user can't map to their code.
Contrived input -- `condition` is a single expression, so you'd need the
`print(x) or cond` idiom -- but it's strictly worse than before, and one line
fixes all four call sites: `f_globals.get('__name__', '<unknown>')` at
`replace_print.py:75`.
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -103,21 +111,73 @@ def __init__(
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, _ = DocumentFactory.open_document(result_uri)
+ return Table(list(document.get()))
+
+ 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 at consume time because at
+ # consume
+ # time THIS worker's own materialization reader is still streaming its
+ # input; issuing a second iceberg/S3 read from this thread while that
+ # reader iterates a lazily-pinned snapshot of a doc that region
+ # re-execution drops and recreates makes the reader fail with S3
+ # "Access Denied" (MinIO's answer for a deleted key). 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
+ executor.attach_loop_table(self._read_loop_input_table())
+ # A Loop End has exactly one input port (port 0); the generated
Review Comment:
This fact is doing more work than the comment gives it credit for. It's
cited here only to justify passing `0`, but the single input port is also what
makes `_process_end_channel` a safe place to read at all.
The chain: `_process_end_channel` is dispatched only from the
`is_ecm_aligned` branch of `_process_ecm`; EndChannel is emitted
`PORT_ALIGNMENT`; `is_ecm_aligned` computes alignment per port, so it fires
only once every channel on that port has delivered it; and each reader emits
its EndChannel only after its iterator has run out. One input port, therefore
every reader is finished before `_read_loop_input_table` runs.
With a second input port that breaks -- alignment is per port, so the
consume would fire at the first port's alignment while the other port's readers
were still streaming, silently putting back exactly the overlap this PR
removed, with no failing test. Worth one clause here, or an assert on the port
count.
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -103,21 +111,73 @@ def __init__(
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, _ = DocumentFactory.open_document(result_uri)
+ return Table(list(document.get()))
+
+ 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 at consume time because at
+ # consume
Review Comment:
The reword stranded a line -- this reads "...because at / consume / time
THIS worker's own materialization reader..." with a bare `consume` hanging off
the end.
While you're here: the diagnosis in this comment is stated as settled fact,
but nothing in the diff distinguishes "a second concurrent read" from "the doc
dropped and recreated under a live reader" -- the second would break the reader
with or without your read. The deferral demonstrably removes reader/read
concurrency, which is real; that it was the cause is still the working
hypothesis. Worth phrasing it that way so the next person debugging a
recurrence doesn't take it as ruled out.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
