Xiao-zhen-Liu commented on code in PR #5700:
URL: https://github.com/apache/texera/pull/5700#discussion_r3561096467
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -87,19 +98,62 @@ 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.
+ uri = self.context.loop_start_state_uris.get(self._loop_start_id)
+ if not uri:
+ raise RuntimeError(
+ f"no loop-back state URI configured for LoopStart "
+ f"'{self._loop_start_id}' "
+ f"(have: {sorted(self.context.loop_start_state_uris)})"
+ )
+ coordinator_interface.jump_to_operator_region(
Review Comment:
There's still no cap on iterations — a `condition` that never turns false
loops forever. Unlike a UDF `while True`, each iteration re-schedules the
region through the controller, recreates the workers, and writes a fresh
iceberg state file, so a runaway loop is repeated controller, worker, and
storage work, not just a spinning thread. I know this was declined before;
worth at least a configurable cap or a periodic log so a stuck loop is visible
and bounded. Your call.
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/PhysicalOp.scala:
##########
@@ -208,6 +208,9 @@ case class PhysicalOp(
// restricting it to only the requiring operator's regions is a possible
// future optimization. Default false.
requiresMaterializedExecution: Boolean = false,
Review Comment:
`requiresMaterializedExecution` replaces the old `isInstanceOf` check. But
the coercion it causes (the forced switch to MATERIALIZED, in
`CostBasedScheduleGenerator.effectiveExecutionMode`) is still silent: a user
who selects PIPELINED on a loop workflow gets MATERIALIZED with no log or
warning. Worth a log line (or a message shown in the UI) when the mode is
overridden, so the UI shows the mode that's actually used.
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -87,19 +98,62 @@ 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.
+ uri = self.context.loop_start_state_uris.get(self._loop_start_id)
+ if not uri:
+ raise RuntimeError(
+ f"no loop-back state URI configured for LoopStart "
+ f"'{self._loop_start_id}' "
+ f"(have: {sorted(self.context.loop_start_state_uris)})"
+ )
+ coordinator_interface.jump_to_operator_region(
+ JumpToOperatorRegionRequest(OperatorIdentity(self._loop_start_id))
+ )
+ writer = DocumentFactory.create_document(uri, State.SCHEMA).writer("0")
Review Comment:
`complete()` guards `condition()` now, which is good — but this state write
isn't guarded. A `put_one`/`close` failure would raise on the main loop thread,
after the jump DCM has already been sent. Worth wrapping it so a write failure
is reported as an error the way the condition path is.
##########
amber/src/test/python/core/models/test_loop_operators.py:
##########
@@ -0,0 +1,394 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Unit tests for the loop runtime: LoopStartOperator and LoopEndOperator.
+
+These exercise the abstract base classes in operator.py that the
+generated `ProcessLoopStartOperator` / `ProcessLoopEndOperator` classes
+extend. The tests use minimal stub subclasses that mirror what
+`LoopStartOpDesc.generatePythonCode` / `LoopEndOpDesc.generatePythonCode`
+emit so the behavior covered here is the same shape that ships at
+runtime.
+
+Coverage:
+ - LoopStart's first-entry state merge into self.state.
+ - 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 multi-iteration loop driven to completion through the operators and the
+ State to_tuple/from_tuple round-trip (TestLoopRunsToCompletion).
+
+loop_counter and the LoopStart jump metadata (LoopStartId) are owned by the
+worker runtime, not these operators -- they ride the StateFrame envelope as
+their own columns (the loop-back write address is setup config, not state) --
+so their handling is covered in test_main_loop.py::TestMainLoop.
+"""
+
+from typing import Iterator, Optional
+
+import pyarrow as pa
+import pytest
+
+from core.models import State, Table, TableLike, Tuple
+from core.models.operator import (
+ _RESERVED_STATE_KEYS,
+ LoopEndOperator,
+ LoopStartOperator,
+)
+from core.models.table import table_from_ipc_bytes, table_to_ipc_bytes
+
+
+# ---------------------------------------------------------------------------
+# Stub subclasses that mirror the generated Python in
+# LoopStart/LoopEnd OpDesc. Keeping them here (rather than reusing the
+# real generator) lets the test pin behavior without spinning up a Scala
+# runtime to produce code.
+# ---------------------------------------------------------------------------
+
+
+class _StubLoopStart(LoopStartOperator):
Review Comment:
These stubs (`_StubLoopStart`/`_StubLoopEnd`) mirror the codegen but aren't
the shipped classes — the real `ProcessLoop*Operator` from `generatePythonCode`
(base64 + `decode_python_template` + exec/eval) is only compiled end-to-end in
the integration test. A fast test that compiles and execs the actual generated
source — including a quote or newline in a user expression — would catch bugs
when the generated code and the stubs get out of sync, without waiting on the
integration job.
##########
amber/src/main/python/core/models/operator.py:
##########
@@ -291,3 +302,152 @@ def process_table(self, table: Table, port: int) ->
Iterator[Optional[TableLike]
time, or None.
"""
yield
+
+
+# ``table`` is seeded into the eval/exec namespaces the loop expressions run
+# in and stripped from every state dict that crosses Loop Start / Loop End,
+# so user code can neither persist nor clobber it. 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).
+_RESERVED_STATE_KEYS: frozenset = frozenset({"table"})
Review Comment:
The reserved set is centralized here, but the literal `"table"` is repeated
at each inject/serialize/read site (332, 380, 432), so it's only partly
centralized — reference the constant at those sites too. Separately: a user
variable literally named `table` is silently dropped by `_strip_reserved`
rather than flagged, so the value quietly disappears. Worth raising on a
collision instead.
##########
amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala:
##########
@@ -0,0 +1,232 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.engine.e2e
+
+import com.twitter.util.Duration
+import org.apache.pekko.actor.{ActorSystem, Props}
+import org.apache.pekko.testkit.{ImplicitSender, TestKit}
+import org.apache.pekko.util.Timeout
+import org.apache.texera.amber.clustering.SingleNodeListener
+import org.apache.texera.amber.core.virtualidentity.OperatorIdentity
+import org.apache.texera.amber.core.workflow.{
+ ExecutionMode,
+ PortIdentity,
+ WorkflowContext,
+ WorkflowSettings
+}
+import org.apache.texera.amber.engine.common.AmberRuntime
+import org.apache.texera.amber.engine.e2e.TestUtils.{
+ buildWorkflow,
+ cleanupWorkflowExecutionData,
+ initiateTexeraDBForTestCases,
+ runWorkflowAndReadResults,
+ setUpWorkflowExecutionData,
+ workflowContext
+}
+import org.apache.texera.amber.operator.LogicalOp
+import org.apache.texera.amber.operator.loop.{LoopEndOpDesc, LoopStartOpDesc}
+import org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpDesc
+import org.apache.texera.amber.tags.IntegrationTest
+import org.apache.texera.workflow.LogicalLink
+import org.scalatest.flatspec.AnyFlatSpecLike
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Outcome, Retries}
+
+import scala.concurrent.duration.DurationInt
+
+/**
+ * End-to-end loop tests: run a real TextInput -> LoopStart -> LoopEnd
workflow
+ * through the engine (controller + Python workers + the DCM back-jump and
+ * region re-execution) and assert both that it terminates AND that it ran the
+ * expected number of iterations.
+ *
+ * Termination alone is too weak: a counter bug that still terminated (e.g.
+ * off-by-one) would pass. So each test asserts the LoopEnd's MATERIALIZED
+ * result-table row count, read from iceberg after the run. LoopEnd is an
+ * identity pass-through on data, so the rows it materializes equal the rows
+ * that flowed through it: a single (outermost) LoopEnd accumulates every
+ * iteration (3 for the single loop; 9 for the terminal outer LoopEnd of the
+ * 3x3 nested loop), and an inner LoopEnd resets once per outer iteration (so
+ * 3, not 9).
+ *
+ * NOTE: the cumulative `ExecutionStatsUpdate` output count is NOT usable as
+ * the iteration count here. A loop region's workers are recreated on every
+ * `JumpToOperatorRegion` re-execution (each iteration spawns a fresh worker
+ * actor), so the per-logical-op output statistic reflects only the final
+ * iteration's worker and does not accumulate. The iceberg result, by
+ * contrast, persists across iterations (the scheduler reuses a LoopEnd's
+ * output document via its output port's `reuseStorage` flag), so it is
+ * the reliable signal.
+ *
+ * Tagged @IntegrationTest because it spawns Python workers; routed to the
+ * `amber-integration` CI job.
+ */
+@IntegrationTest
+class LoopIntegrationSpec
+ extends TestKit(ActorSystem("LoopIntegrationSpec",
AmberRuntime.pekkoConfig))
+ with ImplicitSender
+ with AnyFlatSpecLike
+ with BeforeAndAfterAll
+ with BeforeAndAfterEach
+ with Retries {
+
+ override def withFixture(test: NoArgTest): Outcome =
+ withRetry { super.withFixture(test) }
+
+ implicit val timeout: Timeout = Timeout(5.seconds)
+
+ // Unique per-suite id so the seeded user/workflow/version/execution rows
+ // (and the context's workflow/execution ids) don't collide with the other
+ // integration suites running against the shared test database (#5888).
+ private val specId = 5
Review Comment:
`specId = 5` collides with `MultiRegionWorkflowIntegrationSpec` (also 5),
and both run in the amber-integration job. `setUpWorkflowExecutionData` /
cleanup key the user/workflow/execution rows only on this int, with no
per-suite unique value, so the two suites share rows — which is what the
comment here says the unique id is meant to prevent. Give this spec an unused
id (e.g. 6).
##########
amber/src/test/python/core/runnables/test_main_loop.py:
##########
@@ -1360,7 +1368,9 @@ def process_state(state: State, port: int) -> State:
monkeypatch.setattr(
Review Comment:
`test_process_state_can_emit_multiple_states` looks identical to
`test_process_state_can_emit_consecutive_states` (line 1142) — same executor,
monkeypatches, inputs, and assertions; they differ only in a type annotation.
One can go, or make them actually differ.
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -87,19 +98,62 @@ 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.
+ uri = self.context.loop_start_state_uris.get(self._loop_start_id)
+ if not uri:
+ raise RuntimeError(
+ f"no loop-back state URI configured for LoopStart "
+ f"'{self._loop_start_id}' "
+ f"(have: {sorted(self.context.loop_start_state_uris)})"
+ )
+ coordinator_interface.jump_to_operator_region(
+ JumpToOperatorRegionRequest(OperatorIdentity(self._loop_start_id))
+ )
+ writer = DocumentFactory.create_document(uri, State.SCHEMA).writer("0")
+ # The back-edge fires only after the matching LoopEnd consumed at
+ # loop_counter == 0, so the next iteration's input starts at depth 0.
+ writer.put_one(executor.state.to_tuple(0))
+ writer.close()
+
def complete(self) -> None:
"""
Complete the DataProcessor, marking state to COMPLETED, and notify the
coordinator.
"""
# flush the buffered console prints
self._check_and_report_console_messages(force_flush=True)
- self.context.executor_manager.executor.close()
+ coordinator_interface = self._async_rpc_client.coordinator_stub()
+ executor = self.context.executor_manager.executor
+ if isinstance(executor, LoopEndOperator):
+ # condition() evaluates a user-supplied expression on this main
+ # loop thread. A UDF error on the data path is caught and
+ # reported (DataProcessor._executor_session); mirror that here so
+ # a bad condition (a typo, an undefined name) surfaces as an
+ # operator-facing error and pauses the worker, instead of killing
+ # the thread through run()'s @logger.catch(reraise=True).
+ try:
+ should_jump = executor.condition()
+ except Exception as err:
Review Comment:
This except block re-implements the exception-reporting sequence
`DataProcessor._executor_session` already does (`logger.exception` ->
`set_exception_info` -> error console message). Worth a shared helper so the
two stay in sync.
--
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]
