Fixed #9387: Add State types for tasks and DAGs

[andrewgodwin]

This adds TaskState and DagState enum types that contain all possible states, makes all other core state constants derive their values from them, and adds a couple of initial type hints that use the new enums (with the plan being that we can add signficantly more later).

closes: #9387

[uranusjr]

Switching TaskState and DagState to actually be an enum, and have to go rewrite a large chunk of core code and potentially break compatability with many providers since it'll change the values of all the states to be of type EnumValue.

Would it help to initially make them a StrEnum, and slowly convert the type hints to take an enum value instead of str?

[dstandish]

it's possible to have a stringy enum class, did you explore that?

see here https://stackoverflow.com/a/63028809/2761682

one concern i have with this PR is the creation of a TaskState class.

there has been periodic discussion on mailing list re adding support for stateful tasks and of late some renewed interest in the topic. adding DagState and TaskState classes would potentially be in conflict with that. if we do create stateful tasks, i think that TaskState as the model through which we store state (i.e. an artifact representin the task's state) would be a better usage of the word than an enum representing the execution state.

i should add, it would probably be more accurate to describe these as TaskInstanceState and DagRunState -- the task and dag themselves don't actually have states of any kind at this time. but the same concern applies.

if you want to make this an enum, why can't you just keep the single State class and make it enum, or enum-like?

[uranusjr]

if you want to make this an enum, why can't you just keep the single State class and make it enum, or enum-like?

I think that’d take away most of the reason of having an enum class. The main advantage of enums is to prevent invalid values from being passed (they’d result in a typing or runtime error). Combining all the possible state values into one enum means not all state values make sense for all contexts (some states are impossible on task instances, for example), and kind of defeats the purpose.

[ashb]

if you want to make this an enum, why can't you just keep the single State class and make it enum, or enum-like?

I think that’d take away most of the reason of having an enum class. The main advantage of enums is to prevent invalid values from being passed (they’d result in a typing or runtime error). Combining all the possible state values into one enum means not all state values make sense for all contexts (some states are impossible on task instances, for example), and kind of defeats the purpose.

Exactly this -- there are some states that only apply to TaskInstances, but that don't apply to DagRuns or (Base)Job classes -- "upstream_failed" for instance (there are more).

[ashb]

there has been periodic discussion on mailing list re adding support for stateful tasks and of late some renewed interest in the topic. adding DagState and TaskState classes would potentially be in conflict with that. if we do create stateful tasks, i think that TaskState as the model through which we store state (i.e. an artifact representin the task's state) would be a better usage of the word than an enum representing the execution state.

I think the "stateful" tasks doesn't conflict with this -- Airflow essentially has a state-machine (if a very poorly defined one) for what states a TI/DR can be in, so adding a whole new state is more work than just changing the value in the DB column, so by making this an explicit enum class it would hopefully make it clearer that you can't make up your own values to store in ti.state and expect it to Just Work.

[dstandish]

I think the "stateful" tasks doesn't conflict with this -- Airflow essentially has a state-machine (if a very poorly defined one) for what states a TI/DR can be in, so adding a whole new state is more work than just changing the value in the DB column, so by making this an explicit enum class it would hopefully make it clearer that you can't make up your own values to store in ti.state and expect it to Just Work.

I don't mean that the two concepts conflict just that I would like to reserve the name TaskState and DagState, potentially, for use with "stateful" tasks.

just trying to think if there's a name we could use for the execution states enum which would not clobber that.

but, maybe owing to the fact that there already is an "execution state" concept, if "stateful tasks" is ever implememted it might be prudent to use the name "watermark" or "artifact" instead.

But I think potentially TaskState could be a good name for a stateful tasks framework, and if we called this enum something like TaskRunState or TaskExecutionState -- or perhaps more accurately TaskInstanceExecutionState there would be no conflict.

Another idea would be to nest the different execution state enums under a States class like States.dag_run.RUNNING etc

[ashb]

I probably would have make State a str-enum class like we did here

airflow/airflow/utils/types.py

Lines 23 to 39 in e4deb0f

	class DagRunType(str, enum.Enum):
	"""Class with DagRun types"""
	BACKFILL_JOB = "backfill"
	SCHEDULED = "scheduled"
	MANUAL = "manual"
	def __str__(self) -> str: # pylint: disable=invalid-str-returned
	return self.value
	@staticmethod
	def from_run_id(run_id: str) -> "DagRunType":
	"""Resolved DagRun type from run_id."""
	for run_type in DagRunType:
	if run_id and run_id.startswith(f"{run_type.value}__"):
	return run_type
	return DagRunType.MANUAL

Is there any downside to doing that?

[andrewgodwin]

Not at all - I was not familiar with the ability to add arbitrary type bases to Enum! I'll try that, it should be significantly better.

[andrewgodwin]

I've reworked this to use a str-type Enum, but the enum values are still odd - calling str() on them outputs the name of the value as TaskState.FAILED, not just failed. I don't think the code is going to like it.

I suspect it's going to massively blow up the test suite, but I want to push it up and run here just to be sure.

[dstandish]

No interest in calling it TaskExecutionState instead of TaskState? 🙂

[andrewgodwin]

No interest in calling it TaskExecutionState instead of TaskState? 🙂

I'd rather we call it TaskInstanceState honestly, but right now I'm just trying to see if this will even work type-wise :)

[dstandish]

I'm just trying to see if this will even work type-wise

Totally makes sense :)

I'd rather we call it TaskInstanceState honestly

Let me offer an argument for not this.

It's possible to use xcom as a store of task instance state, because it is at the grain of task instance (task + execution date).

So we can imagine that XCom could have reasonably been called TaskInstanceState.

What this enum represents is something very different. It's the TaskInstance's execution state -- not a way of storing state data for a task instance.

That's why I like the idea of adding the word execution or run.

Separately, I think we could justify going without the word instance because this enum represents possible values for task execution states in the abstract, and it is maybe irrelevant that the when a task gets combined with an execution date it is called a task instance.

[andrewgodwin]

I get your point, but we also can't escape that it is the possible values of the state column on TaskInstance. I agree the naming that's there already isn't amazing, but there's also something to be said for the principle of least surprise :)

[ashb]

@andrewgodwin ping -- when you have a few spare cycles would be good to revisit this PR

[ashb]

Code looks good now -- @dstandish Are you happy with these names (TaskState and DagState)?

[dstandish]

Honestly do not like them for the same reasons, namely that I think there wants to be a mechanism of state persistence in airflow and these names could collide with that

But I concede the alternative names for the enums are maybe not the most appealing e.g. TaskExecutionState or DagExecutionState (though at the moment they seem prefereable to me).

And if anything materializes re state persistence mechanism, I'm sure we'll be able to figure something out. Perhaps that would mean calling it Watermark or Artifact. Or ProcessState. Or perhaps there would not need to be a class like that.

Anyway carry on, if you guys think it's the right idea, it's probably not that bad of an idea :)

[andrewgodwin]

Yeah, I personally think the boat has sailed on reclaiming the word "State" in the codebase right now - these enums reflect the possible values of the state columns on their models, so I think the name is "good enough" given that.

[dstandish]

these enums reflect the possible values of the state columns on their models

By that logic shouldn't these enums be TaskInstanceState and DagRunState? There actually is no Task model and I think DAG doesn't have a state.

[andrewgodwin]

That is an excellent point. Happy to change them if we all agree.

[dstandish]

just noticed few small little docstring things while looking it over

i'm deferring to you and others on the class naming question

[andrewgodwin]

I decided to rename the classes under the banner of being more explicit.

[uranusjr]

LGTM

[uranusjr]

Huh, surprised this type hint is needed.

[andrewgodwin]

It may not be, I just like adding explicit type hints where I think it helps the reader of the code understand what's going on.

[potiuk]

Yep. I also often do that even if not technically needed to make it explicit. I think code should be much more optimised for reader's time than writer's time to write it ;).

[github-actions]

The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.

[ashb]

D'oh, we had two changes in flight -- this one and @ephraimbuddy's #16401, so we'll need to add Queued state to DagRuns too