Re: [I] Allow task to depend on specific tasks from previous dag_run [airflow]

["yuseok89 (via GitHub)" <gi...@apache.org>]

yuseok89 commented on issue #60328:
URL: https://github.com/apache/airflow/issues/60328#issuecomment-5322322206

   Before adding a new parameter, I wanted to check whether the same behaviour can be had with what's already there, and see whether it covers your case.
   
   In the example from the issue, C needs `{C, D, E}`, which is exactly "C itself plus everything transitively downstream of C". `wait_for_downstream=True` does almost that. It turns on `depends_on_past` as well and then additionally waits on downstream tasks, but only immediately downstream ones. So setting it on C covers `{C, D}` and misses E.
   
   `downstream_task_ids` is just a set, and adding a transitive edge doesn't change execution order (the chain already enforces it), so a single edge closes that gap.
   
   ```python
   A >> B >> C >> D >> E
   C >> E   # transitive edge, same execution order, just widens C's direct downstream
   
   C = PythonOperator(
       task_id="C",
       wait_for_downstream=True,   # implies depends_on_past=True
       ...
   )
   ```
   
   `wait_for_downstream` now covers `{D, E}` and the implied `depends_on_past` covers `C`, giving `{C, D, E}`. The first Dag run isn't blocked either, since the check passes when there's no previous run.
   
   Would this work in your actual pipeline, or does it miss something? If it covers the case, documenting the pattern under `wait_for_downstream` might be more useful than a new argument. If it doesn't, knowing why would sharpen the case for a first-class parameter.


-- 
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