depends_on_past is ignored when upstream is skipped

[hussein-awala]

Apache Airflow version

2.3.4

What happened

If we have an hourly dag of 2 tasks T1 and T2 (T1 >> T2), and T2 has depends_on_past=True, if at the run of 10:00 T1 is skipped, T2 is directly marked as skipped too without waiting the task of the previous run 09:00, then in the next run 11:00, T2 can be executed where T2 of its previous run is skipped.

What you think should happen instead

If the task has depends_on_past=True, we should not change its state before that the same task of the previous run is marked as succeeded or skipped.

How to reproduce

Here is a simple dag which can help to reproduce the problem:

from datetime import datetime

from airflow import DAG
from airflow.exceptions import AirflowSkipException
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator

with DAG(
        dag_id='dag',
        start_date=datetime(2022, 1, 1),
        schedule_interval=None,
) as dag:
    def t1_callable(**context):
        dag_run_conf = context["dag_run"].conf
        if dag_run_conf.get("skip_t1"):
            raise AirflowSkipException()

    t1 = PythonOperator(
        task_id="T1",
        python_callable=t1_callable
    )
    t2 = BashOperator(
        task_id="T2",
        bash_command="sleep {{dag_run.conf.get('sleep_seconds', 1)}}",
        depends_on_past=True,
    )
    t1 >> t2

Then using Airflow CLI:

airflow dags trigger dag --conf '{"sleep_seconds":1000}'
airflow dags trigger dag --conf '{"skip_t1":true}'
airflow dags trigger dag --conf '{"sleep_seconds":10}'

Operating System

Debian GNU/Linux

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

Anything else

In some cases max_active_runs can help to avoid this problem, but we have dags with tasks not dependent on past, which we prefer to run ASAP and keep the dag running waiting the tasks which depend on past runs.

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template!

[potiuk]

Hmmm I think everything is as expected.

1. when you manually set task on 10.00 to "skipped" - it will be skipped without taking "depends_on_past" into account. "Depends_on_past" only prevents the tasks from RUNNING when the same task in previous instance is not "succeeded" or "skipped". But when you manually force it to "skip" this is not done. If you would like it to "wait" for previous task to run, you would have to have a new state "waiting for skipping" or smth. I.e. you would like to have a state when we know that the task should be skipped immediately when it is eligible to run. Airfow has no such state and it's quite doubtful if it will ever have one

2. Then the task at 11.00 will only look at the "last previous", not "all previous" tasks and it will see tha the previous task is skipped, so it will run immediately.

All is fine, as expected and perfectly reflects the current state of Airflow implementation, so this is not a bug.

What you likely would like to get instead is a new feature for example "depend_on_all_past_not_runnning" or similar, where you should check if there is ANY past task that is currently not "completed" and "queued" or "running". But I think this one is not needed. It is quite a bit more complex to implemented, I think and it might be a bit more "predictable" in terms of execution sequence, but probably much better, simpler and something that you can implement even now, is to assign a separate pool for your task with size = 1. This will make sure that only one task instance can run at a time. This is PROBABLY what you really want to achieve. There is no guarantee on sequence of execution - so if you run 5 of those instances one after the other then, they will execute in a "random" sequence, but In most cases of scheduled run, they should run in a proper sequence as past task will be already running while the future task is scheduled.

Converting to a discussion.

[hussein-awala]

I just created a new draft PR which add a new conf for task skipping, when it is false, and it is decided to skip a task, the task will stay in None state waiting the past depends to be met. In this case no need to check all the past depends because each one ensure that the previous one is met.

I didn't add unit tests yet, I prefer to validate the concept with you before adding them. Can you please check it?

[dstandish]

Can you provide an example of a practical use case?

[hussein-awala]

Our use case

We are using Airflow to manage our data lakehosue (over 300 tables), we have a dag per (table, operation), in total 1200 dags. One of this operation is ingress. The dag consists of two tasks:

• A sensor on AWS Glue which check if the partition (execution date, execution hour) exists in Glue Metastore.
• A task to submit as Spark job to load the parquet data in our apache Hudi lakehouse.

When the sensor doesn't find a partition for the current execution datetime and find a one for the next execution datetime, it raise AirflowSkipException to skip running the spark job where there is no data to process (custom soft fail).
We need to process the data in a chronological order for some business needs, for that we set depends_on_past=True for the two tasks.

What is the problem we are facing?

The problem we have is when the sensor is marked as skipped, the Spark task is marked as skipped too without waiting the previous task to be done, and the next task can be executed and terminated before a past task.

What do we expect from Airflow?

We expect that the Depends on past forces Airflow tasks to be executed and terminated in chronological order.

The solution we have tested

1. Set max_active_runs to 1: it can't solve the problem because when a run fails, and the next one is marked as skipped, the next runs will be executed normally.
2. Avoid skip the sensor and run the Spark task anyway: if there is no data to process, we create a Spark job which reserve some resources and may scale up the cluster for nothing. And it logs OpenLineage metrics without touching the dataset.
3. Skip the sensor but change the Spark operator execution code to check the sensor state: it adds complexity to our operator code, and we have different operators for different operations, so a lot of code to manage.

We believe that this problem should be solved in Airflow, for that we decided to contribute and add the new feature which keep the task in None state waiting the previous task, before marking it as skipped.