aglinxinyuan commented on code in PR #6971:
URL: https://github.com/apache/texera/pull/6971#discussion_r3700739230
##########
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:
Fixed in 25d6ace, exactly as you suggested — `f_globals.get('__name__',
'<unknown>')`. Reproduced your two one-liners first; both now capture instead
of raising.
Agreed it's strictly worse than before for `condition` specifically, and the
blast radius was wider than the new code: on main the same crash already hit a
`print()` in the Loop End `update` and the Loop Start `output` (both ran under
`_executor_session`'s `replace_print`), so this also fixes those.
##########
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:
Right — both tests exercised a path the shipped operator never takes.
Rewritten in 25d6ace: `_PrintingLoopEnd` now mirrors
`LoopEndOpDesc.generatePythonCode` — `condition` goes through
`self.eval_condition("print('...') or False")` and `process_state` through
`self.run_update("print('...')\ni += 1", state)` — so the print executes in an
eval/exec frame with no `__name__`, which is where the capture used to crash.
Verified both go red (with the exact `KeyError: '__name__'` from
`replace_print.py`) when the fix is reverted.
##########
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:
Both points taken — reworded in 25d6ace. The comment now says what the flag
actually does: the consume runs the update exactly once either way (and
`run_update` seeds from the incoming copy, so identical copies would even be
idempotent); what the flag pins is which copy is stashed — first-wins instead
of each later arrival silently overwriting `_pending_loop_state`.
On the identity claim: kept as the normal case but no longer load-bearing —
the comment now states that rewriting the loop state in a branching body is not
supported (which copy arrives first is scheduling-dependent either way), so the
first is taken deliberately.
##########
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:
Done in 25d6ace — the consume now clears the flag right where it takes the
stash, with your port-alignment argument as the stated reason (nothing more can
arrive on the port after the aligned EndChannel, so every duplicate is already
in). The field comment now derives per-execution from construction instead of
from the scheduler's worker-per-iteration behavior, and the fan-in test pins
the re-arm (`_loop_state_consumed is False` after the consume).
##########
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:
Hoisted in 25d6ace — `_process_end_channel` now holds the region immediately
after `process_input_state()`, before the consume block, so an already-reported
`produce_state_on_finish` error is not followed by a storage read, a user
`update`, or a second stacked error. The consume keeps its own hold below for
its own failures.
##########
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:
Added the chain to the comment in 25d6ace: EndChannel is PORT_ALIGNMENT
(dispatched once every channel of the port has delivered it), each reader emits
its EndChannel only after its iterator is exhausted, so with one input port
every reader has finished before the read — and a second input port would break
that silently, since alignment is per port. Went with the clause rather than an
assert: the port map lives behind `InputManager`'s private `_ports` and I'd
rather not reach into it from the consume path; if Loop End ever grows a second
input port, the warning now sits directly on the line that stops being safe.
##########
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:
Cast added in 25d6ace — `_read_loop_input_table` now keeps the schema
`open_document` returns and runs `cast_to_schema` on every tuple, same as
`input_port_materialization_reader_runnable`, with a comment saying why the two
readers of one doc must not normalize differently. The read-URI unit test now
also asserts every tuple was cast to the doc's schema.
##########
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:
Both fixed in 25d6ace. The wording reads through cleanly now, and the
diagnosis is framed as what it is: removing the overlap made the 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 a recurrence should be treated as new evidence.
--
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]
