`wait_for_past_depends_before_skipping` does not seem to have any effect
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 7h
- Merged PRs (30d)
- 484
Description
### Apache Airflow version
Airflow 3.1.0
### What happened?
I am part of a data engineering team. We use airflow to orchestrate our data ingestion pipelines. At a high-level, they are made of:
- An ingestion task group, that moves data from a source system to our S3 storage
- Validation tests
- Processing jobs that clean up and prepares the data for use by downstream consumers
- Validations tests on the processed data
The processing jobs run in Spark. To avoid overhead, I have been looking at using Airflow's skipping mechanisms to skip the rest of the pipeline if there is no data to be ingested.
However, we need the processing jobs to run sequentially. For this purpose, we use `depends_on_past`. I then stumbled upon [the PR](https://github.com/apache/airflow/pull/27710) introducing `wait_for_past_depends_before_skipping`, which seems like it would allow me to skip tasks while still preserving the dependencies on past runs. However, the flag does not seem to have an effect.
### What you think should happen instead?
The flag should behave as advertised in the base PR and in the docs.
### How to reproduce
```python
from airflow import DAG
from airflow.models import Param
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.operators.python import ShortCircuitOperator
from airflow.sdk import get_current_context
def check_continue():
ctx = get_current_context()
skip_downstream = ctx["params"]["skip"]
return not skip_downstream
with DAG(
dag_id="test_airflow_skip",
schedule=None,
params={
"skip": Param(default=False, type="boolean"),
},
) as dag:
depends_on_past = EmptyOperator(
task_id="depends_on_past",
depends_on_past=True,
wait_for_past_depends_before_skipping=True,
)
skip = ShortCircuitOperator(
task_id="skip",
python_callable=check_continue,
)
skip >> depends_on_past
```
1. Launch a run
2. Then manually set the state of the `depends_on_past` task to `failed`
3. Then launch a run with the `skip` parameter enabled: `depends_on_past` will be skipped when it shouldn't be.
### Operating System
Ubuntu 22.04.5 LTS
### Versions of Apache Airflow Providers
_No response_
### Deployment
Docker-Compose
### Deployment details
_No response_
### Anything else?
_No response_
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
Contributor guide
Research direction
Reproduce the behavior with the provided DAG, EmptyOperator, ShortCircuitOperator, and depends_on_past settings, then compare the flag's implementation with PR 27710 and the referenced documentation. Done means a failed previous task prevents the current task from being skipped when wait_for_past_depends_before_skipping is enabled, with coverage for the supplied sequence.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- docker-compose, python
- Domain
- backend, data-engineering
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 42/100