kaxil commented on code in PR #74773:
URL: https://github.com/apache/airflow/pull/74773#discussion_r4319948287
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py:
##########
@@ -2696,7 +2696,7 @@ def complete():
assert gate.state == State.SUCCESS
assert sorted(
ti.region_index for ti in dr.get_task_instances(session=session)
if ti.task_id == gate.task_id
- ) == [0, 1]
+ ) == [1, 2]
def test_loop_gate_completion_uses_its_pinned_limit(self, client, session,
running_loop_gate, dag_maker):
Review Comment:
The `ti.region_index == 1` filter in this test now picks up the first pass
(gate and terminal) instead of the successor, so `len(successors) == 2` and the
version check pass whatever the server does with the pinned limit. Should be
`== 2`.
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py:
##########
@@ -151,6 +151,21 @@ def
test_previous_iteration_xcom_uses_retained_predecessor(client, loop_xcoms, s
assert client.head(url, params=params).headers["Content-Range"] ==
"map_indexes 1"
+def test_previous_iteration_of_the_first_pass_resolves_to_nothing(client,
loop_xcoms, authenticate_as):
Review Comment:
This also passes with the old `< 0` guard: pass 1 then looks up pass 0,
which never exists, so it 404s either way. To pin "stops before looking for
one", assert the producer query never runs (`capture_orm_selects`, as elsewhere
in this file), or trim the docstring to the 404.
##########
airflow-core/docs/authoring-and-scheduling/loops.rst:
##########
@@ -290,9 +290,10 @@ first iteration, then each previous result plus one.
:start-after: [START mapped_loop]
:end-before: [END mapped_loop]
-The loop iteration and mapped position are separate "coordinates". In this
-example, each iteration has mapped instances 0 and 1, so mapped instance 1 in
-iteration 0 is distinct from mapped instance 1 in iteration 1.
+The loop iteration and mapped position are separate "coordinates", and they do
Review Comment:
"Rerun part of an iteration" further down (L389 and L396) still counts from
zero: "iterations 0 through 4" and "Iterations 0 and 1 remain untouched". With
this change that becomes "1 through 4" and "Iteration 1 remains untouched".
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py:
##########
@@ -2499,17 +2499,17 @@ def clear_again():
@pytest.mark.parametrize(
("decision", "state", "max_iterations", "conditional", "rejected",
"passes"),
[
- ("continue", State.SUCCESS, 3, False, False, [0, 1]),
- ("stop", State.SUCCESS, 3, False, True, [0]),
- (None, State.SUCCESS, 3, False, True, [0]),
- ("continue", State.FAILED, 3, False, False, [0]),
- ("continue", State.SKIPPED, 3, False, False, [0]),
- ("stop", State.SUCCESS, 1, False, False, [0]),
- ("continue", State.SUCCESS, 1, False, True, [0]),
- ("stop", State.SUCCESS, 3, True, False, [0]),
- ("stop", State.SUCCESS, 1, True, False, [0]),
- ("continue", State.SUCCESS, 1, True, True, [0]),
- (None, State.FAILED, 1, True, False, [0]),
+ ("continue", State.SUCCESS, 3, False, False, [1, 2]),
Review Comment:
With this row now `[1, 2]`, the `if mapped and passes == [0, 1]:` block
further down (L2584) no longer matches any row, so the check that the next
pass's mapped task starts as a `-1` placeholder and then expands silently
stopped running. It needs `passes == [1, 2]` and `parent_region_index == 2`
there.
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py:
##########
@@ -54,7 +54,7 @@ class LoopIterationResponse(BaseModel):
"""A task execution's position in an enclosing loop."""
loop_id: str
- iteration: int
+ iteration: int = Field(ge=1, description="Position in the enclosing loop,
counting from one.")
Review Comment:
3.4.0b1 already created loop passes at index 0 and nothing renumbers them on
upgrade, so on a DB carried over from b1 a pass-0 row fails `ge=1` when the
response is serialized, and the TI list or gantt for that run returns a 500
instead of rendering. `LoopContext.index` has the same issue for a cleared
pass-0 TI at `ti_run`. Only beta users can hit this, but would it be simpler to
keep the bound on the `?iteration=` query param and leave the response models
unconstrained? Otherwise a line in the release notes for b1 testers.
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -1107,6 +1107,10 @@ class TaskInstance(Base, LoggingMixin, BaseWorkload):
task_id: Mapped[str] = mapped_column(StringID(), nullable=False)
dag_id: Mapped[str] = mapped_column(StringID(), nullable=False)
run_id: Mapped[str] = mapped_column(StringID(), nullable=False)
+ # The slot this instance fills in its region, so its meaning follows the
region's node: a loop
+ # region numbers its iterations from one, a mapped expansion numbers its
slots from zero, and -1
+ # is the placeholder for a mapped task that has not been expanded yet.
Code that reads the column
Review Comment:
Nit: plain tasks outside any region also store `-1` (they get `(-1,)` in
`dagrun.py`), and that's the most common case. Can we mention it here too?
--
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]
