You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "seanghaeli (via GitHub)" <gi...@apache.org> on 2025/07/08 00:10:49 UTC

[PR] Move UPSTREAM_FAILED to Terminal State and Add State Validation Tests [airflow]

seanghaeli opened a new pull request, #53000:
URL: https://github.com/apache/airflow/pull/53000

   PR for #52999. Thoughts on this change? Motivated from #52722, wanted to make this a separate PR for it since its scope it quite different.


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Move UPSTREAM_FAILED to Terminal State and Add State Validation Tests [airflow]

Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on PR #53000:
URL: https://github.com/apache/airflow/pull/53000#issuecomment-3053287115

   @o-nikolas Tests passing, ready for review.


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Move UPSTREAM_FAILED to Terminal State and Add State Validation Tests [airflow]

Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on code in PR #53000:
URL: https://github.com/apache/airflow/pull/53000#discussion_r2217056945


##########
airflow-core/tests/unit/utils/test_state.py:
##########
@@ -76,3 +76,39 @@ def test_dagrun_state_enum_escape():
         assert rows[0].state == "queued"
 
         session.rollback()
+
+
+class TestTaskInstanceStates:

Review Comment:
   I think it's good for them to enforce synchronization between `TaskInstanceStates`, `TerminalTIState`, `IntermediateTIState`, `success_states`, and `failed_states`. The reason this PR is here is because in places like this:
   
   https://github.com/apache/airflow/blob/d4473555c0e7022e073489b7163d49102881a1a6/providers/amazon/src/airflow/providers/amazon/aws/sensors/mwaa.py#L99-L100
   
   It is best to iterate instead of hardcoding. Being certain that the success_states are in sync with all terminal states (in case more are added in the future) is useful.



-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Move UPSTREAM_FAILED to Terminal State and Add State Validation Tests [airflow]

Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #53000:
URL: https://github.com/apache/airflow/pull/53000#discussion_r2217404675


##########
airflow-core/tests/unit/utils/test_state.py:
##########
@@ -81,6 +81,23 @@ def test_dagrun_state_enum_escape():
 class TestTaskInstanceStates:
     """Test suite for validating task instance state relationships."""
 
+    def test_failed_and_success_states_union_equals_terminal_states(self):
+        """Test that union of failed_states and success_states equals TerminalTIState values."""
+        # Get all terminal state values
+        terminal_state_values = {state.value for state in TerminalTIState}
+
+        # Get union of failed and success states (convert to string values)
+        failed_success_union = {
+            state.value for state in State.failed_states.union(State.success_states)
+        }.union(
+            {"removed"}
+        )  # Treat removed separately since it doesn't neatly fit into the "success" or "failure" category semantically.ßß

Review Comment:
   ```suggestion
           )  # Treat removed separately since it doesn't neatly fit into the "success" or "failure" category semantically.
   ```



-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Move UPSTREAM_FAILED to Terminal State and Add State Validation Tests [airflow]

Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on PR #53000:
URL: https://github.com/apache/airflow/pull/53000#issuecomment-3049661515

   > @o-nikolas if you don't mind, let me know what you think
   
   I like this change. All the sets that we have for specific terminal states (I think it's just `success` and `failure` states at the moment) should definitely sum up to the full set of terminal states. This makes the set math you're trying to do in your original PR much cleaner, future proof and not hardcoded to specific sets which is all great stuff.
   
   Only concern I'd have is what you mention, the changes propagating in ways we don't expect, but I don't think this is insurmountable and I'd be interested to see hear how many tests failed because of this change. 


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Move UPSTREAM_FAILED to Terminal State and Add State Validation Tests [airflow]

Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on code in PR #53000:
URL: https://github.com/apache/airflow/pull/53000#discussion_r2196295182


##########
airflow-core/src/airflow/utils/state.py:
##########
@@ -213,7 +213,7 @@ def color_fg(cls, state):
     """
 
     success_states: frozenset[TaskInstanceState] = frozenset(
-        [TaskInstanceState.SUCCESS, TaskInstanceState.SKIPPED]
+        [TaskInstanceState.SUCCESS, TaskInstanceState.SKIPPED, TaskInstanceState.REMOVED]

Review Comment:
   1. Implications for `TerminalStateNonSuccess`:
       - In the main branch, I've verified that when calling the `finish` function in `client.py` with state parameter as `UPSTREAM_FAILED`, an error will be thrown. This is due to the following line of the function:
          https://github.com/apache/airflow/blob/ac9e968fe85cf43717d280f303470eb5268ca9ca/task-sdk/src/airflow/sdk/api/client.py#L157
          In which `TerminalStateNonSuccess("upstream_failed")` is problematic
   
         My conclusion then is that nowhere in the main branch is using the `finish` function in this way, otherwise we would be seeing errors.
       - In this PR, looking into a future situation where the `finish` function is called with the state parameter as `UPSTREAM_FAILED`, the `test_task_instance_finish` test verifies that it works:
   https://github.com/apache/airflow/blob/59fa23fd2215354576efeefad77b506449eb2c36/task-sdk/tests/task_sdk/api/test_client.py#L287-L308
       - I've spun up a DAG where there is a task with terminal state `SUCCESS`, and another task with terminal state `UPSTREAM_FAILED`. For both tasks, the `finish` function is not being called (I put an exception within the function and it doesn't get triggered, the state change is being triggered from somewhere else). Additionally, both in the main branch and in this PR, the `end_date` associated with the Task Instance with `UPSTREAM_FAILED` gets set normally, so I don't think there's issues with that specifically.
   2. Adding `REMOVED` to the `success_states` set:
       - This addresses the part of the PR that says `failed + success = terminal`. Currently, without this change, the union of the `success` and `failed` sets does not cover all of the `terminal` states. 
       - I can't reproduce a situation where a Task gets put into the `REMOVED` state since when that happens the DAG version just updates and the missing Task just disappears on the Airflow UI. If you agree that `REMOVED` is not a relevant terminal state, then I can go ahead and remove it from the list of success states and we can still effectively consider all reachable `failed + success` states indeed equal `terminal`.



-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Move UPSTREAM_FAILED to Terminal State and Add State Validation Tests [airflow]

Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #53000:
URL: https://github.com/apache/airflow/pull/53000#discussion_r2195514312


##########
airflow-core/src/airflow/utils/state.py:
##########
@@ -213,7 +213,7 @@ def color_fg(cls, state):
     """
 
     success_states: frozenset[TaskInstanceState] = frozenset(
-        [TaskInstanceState.SUCCESS, TaskInstanceState.SKIPPED]
+        [TaskInstanceState.SUCCESS, TaskInstanceState.SKIPPED, TaskInstanceState.REMOVED]

Review Comment:
   I don't know if this right. The PR description doesn't explain about refactoring to removed state and motivation for it. Please don't do such changes on the fly. This is the very heart of Airflow - doing things wrong here may result in yanking a release.
   
   Given dag versioning, I am not sure what removed state means or if it's even possible to have a task in removed state? in any case best to keep this in a separate PR to not lose focus of separated topics



##########
airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py:
##########
@@ -66,6 +66,7 @@ class TerminalStateNonSuccess(str, Enum):
     FAILED = TerminalTIState.FAILED
     SKIPPED = TerminalTIState.SKIPPED
     REMOVED = TerminalTIState.REMOVED
+    UPSTREAM_FAILED = "upstream_failed"

Review Comment:
   Why hard coded string?



-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Move UPSTREAM_FAILED to Terminal State and Add State Validation Tests [airflow]

Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil merged PR #53000:
URL: https://github.com/apache/airflow/pull/53000


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Move UPSTREAM_FAILED to Terminal State and Add State Validation Tests [airflow]

Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #53000:
URL: https://github.com/apache/airflow/pull/53000#discussion_r2216180393


##########
airflow-core/src/airflow/utils/state.py:
##########
@@ -213,7 +213,7 @@ def color_fg(cls, state):
     """
 
     success_states: frozenset[TaskInstanceState] = frozenset(
-        [TaskInstanceState.SUCCESS, TaskInstanceState.SKIPPED]
+        [TaskInstanceState.SUCCESS, TaskInstanceState.SKIPPED, TaskInstanceState.REMOVED]

Review Comment:
   Conceptually, `REMOVED` as part of `success_states` doesn't make sense to me



-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Move UPSTREAM_FAILED to Terminal State and Add State Validation Tests [airflow]

Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on PR #53000:
URL: https://github.com/apache/airflow/pull/53000#issuecomment-3049644091

   @o-nikolas if you don't mind, let me know what you think


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Move UPSTREAM_FAILED to Terminal State and Add State Validation Tests [airflow]

Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #53000:
URL: https://github.com/apache/airflow/pull/53000#discussion_r2216822995


##########
airflow-core/tests/unit/utils/test_state.py:
##########
@@ -76,3 +76,39 @@ def test_dagrun_state_enum_escape():
         assert rows[0].state == "queued"
 
         session.rollback()
+
+
+class TestTaskInstanceStates:
+    """Test suite for validating task instance state relationships."""
+
+    def test_terminal_and_intermediate_states_union_equals_task_instance_states(self):
+        """Test that union of TerminalTIState and IntermediateTIState values equals TaskInstanceState values."""
+        # Get all terminal state values
+        terminal_state_values = {state.value for state in TerminalTIState}
+
+        # Get all intermediate state values
+        intermediate_state_values = {state.value for state in IntermediateTIState}
+
+        # Get union of terminal and intermediate states
+        terminal_intermediate_union = terminal_state_values.union(intermediate_state_values)
+
+        # Get all TaskInstanceState values, excluding RUNNING since it's not in terminal or intermediate
+        task_instance_state_values = {state.value for state in TaskInstanceState}
+
+        # Add RUNNING state separately since it's defined directly in TaskInstanceState
+        expected_task_instance_values = terminal_intermediate_union.union({"running"})
+
+        assert task_instance_state_values == expected_task_instance_values, (
+            f"TaskInstanceState values ({task_instance_state_values}) "
+            f"does not equal union of TerminalTIState and IntermediateTIState values plus RUNNING "
+            f"({expected_task_instance_values})"
+        )
+
+    def test_no_overlap_between_failed_and_success_states(self):
+        """Test that failed_states and success_states have no overlap."""
+        overlap = State.failed_states.intersection(State.success_states)
+        assert len(overlap) == 0, f"failed_states and success_states should not overlap, but found: {overlap}"
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])

Review Comment:
   ? We don't need this!



-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Move UPSTREAM_FAILED to Terminal State and Add State Validation Tests [airflow]

Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #53000:
URL: https://github.com/apache/airflow/pull/53000#discussion_r2216823652


##########
airflow-core/tests/unit/utils/test_state.py:
##########
@@ -76,3 +76,39 @@ def test_dagrun_state_enum_escape():
         assert rows[0].state == "queued"
 
         session.rollback()
+
+
+class TestTaskInstanceStates:

Review Comment:
   I don't find much value in this test tbh



-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Move UPSTREAM_FAILED to Terminal State and Add State Validation Tests [airflow]

Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on PR #53000:
URL: https://github.com/apache/airflow/pull/53000#issuecomment-3110221413

   > `UPSTREAM_FAILED` change looks good but adding of `removed` to `success_states` does not
   
   In my opinion, `REMOVED` semantically makes sense as a failure state. It represents work that we expect to be completed at the time the DAG is initially called, but then is no longer available when it comes time to execute the task. Would you agree @kaxil ?


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Move UPSTREAM_FAILED to Terminal State and Add State Validation Tests [airflow]

Posted by "seanghaeli (via GitHub)" <gi...@apache.org>.
seanghaeli commented on PR #53000:
URL: https://github.com/apache/airflow/pull/53000#issuecomment-3074768361

   @kaxil Since you created the `IntermediateTIState` and `TerminalTIState` classes, what are your thoughts about moving the Task Instance State `UPSTREAM_FAILED` from `IntermediateTIState` to `TerminalTIState`?


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org