You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2020/03/16 12:36:01 UTC
[GitHub] [airflow] TeddyHartanto opened a new pull request #7735:
[AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
TeddyHartanto opened a new pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735
Before this fix, tasks that are in SUCCESS or SKIPPED state
satisfy the depends_on_past check, but only tasks that are in
SUCCESS state satisfy the wait_for_downstream check. That makes the API
inconsistent and unintuitive.
---
Issue link: WILL BE INSERTED BY [boring-cyborg](https://github.com/kaxil/boring-cyborg)
Make sure to mark the boxes below before creating PR: [x]
- [x] Description above provides context of the change
- [x] Commit message/PR title starts with `[AIRFLOW-NNNN]`. AIRFLOW-NNNN = JIRA ID<sup>*</sup>
- [x] Unit tests coverage for changes (not needed for documentation changes)
- [x] Commits follow "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)"
- [x] Relevant documentation is updated including usage instructions.
- [x] I will engage committers as explained in [Contribution Workflow Example](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#contribution-workflow-example).
<sup>*</sup> For document-only changes commit message can start with `[AIRFLOW-XXXX]`.
---
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/master/UPDATING.md).
Read the [Pull Request Guidelines](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#pull-request-guidelines) for more information.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] ashb commented on a change in pull request #7735:
[AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#discussion_r409397142
##########
File path: tests/models/test_dagrun.py
##########
@@ -552,3 +561,131 @@ def with_all_tasks_removed(dag):
dagrun.verify_integrity()
flaky_ti.refresh_from_db()
self.assertEqual(State.NONE, flaky_ti.state)
+
+ def test_depends_on_past(self):
+ # dag_id = 'test_depends_on_past'
+ # dag = self.dagbag.get_dag(dag_id)
+ # task = dag.tasks[0]
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ # ti1 = TI(task, timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # ti2 = TI(task, timezone.datetime(2016, 1, 2, 0, 0, 0))
+ #
+ # def run_2nd_dagrun():
+ # dag.run(
+ # start_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # end_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # executor=MockExecutor()
+ # )
+ #
+ # # 2nd run of task fails by itself
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SCHEDULED)
+ #
+ # # 2nd run af task fails if 1st run of task failed
+ # ti1.set_state(State.FAILED)
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SCHEDULED)
+ #
+ # # but it works after 1st instance of task is marked as skipped
+ # ti1.set_state(State.SKIPPED)
+ # run_2nd_dagrun()
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SUCCESS)
+ #
+ # # and it also works if 1st instance is success
+ # ti2.set_state(State.NONE)
+ # ti1.set_state(State.SUCCESS)
+ # run_2nd_dagrun()
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SUCCESS)
+ dag_id = 'test_depends_on_past'
+
+ dag = self.dagbag.get_dag(dag_id)
+ task = dag.tasks[0]
+
+ self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+
+ uti1 = TI(task, timezone.datetime(2016, 1, 1, 0, 0, 0))
+ uti2 = TI(task, timezone.datetime(2016, 1, 2, 0, 0, 0))
+
+ for state in State.task_states:
+ uti1.set_state(state)
+ uti2.set_state(State.QUEUED)
+ uti2.run()
+ if state in (State.SUCCESS, State.SKIPPED):
+ self.assertEqual(uti2.state, State.SUCCESS)
+ else:
+ self.assertNotEqual(uti2.state, State.SUCCESS)
+
+ def test_wait_for_downstream(self):
+ # dag_id = 'test_wait_for_downstream'
+ # dag = models.DagBag().get_dag(dag_id)
+ #
+ # TODO: clarify with airflow maintainers -- is the snippet below still true and necessary?
+ # # ti.previous_ti requires a dagrun to exist
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ # upstream, downstream = dag.tasks
+ #
+ # def run_2nd_dagrun():
+ # dag.run(
+ # start_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # end_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # executor=MockExecutor()
+ # )
+ #
+ # # doesn't run if downstream task for previous day has a null state
+ # uti_2 = TI(task=upstream, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ # uti_2.previous_ti.set_state(State.SUCCESS)
+ # self.assertEqual(uti_2.previous_ti.state, State.SUCCESS)
+ # TODO: clarify with airflow maintainers -- I don't think this triggers an AirflowException, does it?
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
+ # uti_2.refresh_from_db()
+ # self.assertEqual(uti_2.state, State.SCHEDULED)
+ #
+ # # doesn't run if downstream task for previous day has a failed state
+ # dti_1 = TI(task=downstream, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # dti_1.set_state(State.FAILED)
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
+ # uti_2.refresh_from_db()
+ # self.assertEqual(uti_2.state, State.SCHEDULED)
+ #
+ # # runs if downstream task for previous day has a skipped state
+ # dti_1 = TI(task=downstream, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # dti_1.set_state(State.SKIPPED)
+ # run_2nd_dagrun()
+ # uti_2.refresh_from_db()
+ # self.assertEqual(uti_2.state, State.SUCCESS)
+ #
+ # # runs if downstream task for previous day has a success state
+ # uti_2.set_state(State.NONE)
+ # dti_1.set_state(State.SUCCESS)
+ # run_2nd_dagrun()
+ # uti_2.refresh_from_db()
+ # self.assertEqual(uti_2.state, State.SUCCESS)
+ dag_id = 'test_wait_for_downstream'
+ dag = self.dagbag.get_dag(dag_id)
+ upstream, downstream = dag.tasks
+
+ # For ti.set_state() to work, the DagRun has to exist,
+ # Otherwise ti.previous_ti returns an unpersisted TI
+ self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+
+ dti1 = TI(task=downstream, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ uti2 = TI(task=upstream, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ uti2.previous_ti.set_state(State.SUCCESS)
+ self.assertEqual(uti2.previous_ti.state, State.SUCCESS)
+
+ for state in State.task_states:
+ dti1.set_state(state)
+ uti2.set_state(State.QUEUED)
+ uti2.run()
+ if state in (State.SUCCESS, State.SKIPPED):
+ self.assertEqual(uti2.state, State.SUCCESS)
+ else:
+ self.assertNotEqual(uti2.state, State.SUCCESS)
Review comment:
This test looks very similar to the one you added in test_taskinstance -- why is one in test_dagrun and the other in test_taskinstance?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] TeddyHartanto commented on a change in pull request
#7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
TeddyHartanto commented on a change in pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#discussion_r409367097
##########
File path: tests/models/test_dagrun.py
##########
@@ -552,3 +561,131 @@ def with_all_tasks_removed(dag):
dagrun.verify_integrity()
flaky_ti.refresh_from_db()
self.assertEqual(State.NONE, flaky_ti.state)
+
+ def test_depends_on_past(self):
+ # dag_id = 'test_depends_on_past'
+ # dag = self.dagbag.get_dag(dag_id)
+ # task = dag.tasks[0]
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ # ti1 = TI(task, timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # ti2 = TI(task, timezone.datetime(2016, 1, 2, 0, 0, 0))
+ #
+ # def run_2nd_dagrun():
+ # dag.run(
+ # start_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # end_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # executor=MockExecutor()
+ # )
+ #
+ # # 2nd run of task fails by itself
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SCHEDULED)
+ #
+ # # 2nd run af task fails if 1st run of task failed
+ # ti1.set_state(State.FAILED)
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SCHEDULED)
+ #
+ # # but it works after 1st instance of task is marked as skipped
+ # ti1.set_state(State.SKIPPED)
+ # run_2nd_dagrun()
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SUCCESS)
+ #
+ # # and it also works if 1st instance is success
+ # ti2.set_state(State.NONE)
+ # ti1.set_state(State.SUCCESS)
+ # run_2nd_dagrun()
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SUCCESS)
+ dag_id = 'test_depends_on_past'
+
+ dag = self.dagbag.get_dag(dag_id)
+ task = dag.tasks[0]
+
+ self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+
+ uti1 = TI(task, timezone.datetime(2016, 1, 1, 0, 0, 0))
+ uti2 = TI(task, timezone.datetime(2016, 1, 2, 0, 0, 0))
+
+ for state in State.task_states:
+ uti1.set_state(state)
+ uti2.set_state(State.QUEUED)
+ uti2.run()
+ if state in (State.SUCCESS, State.SKIPPED):
+ self.assertEqual(uti2.state, State.SUCCESS)
+ else:
+ self.assertNotEqual(uti2.state, State.SUCCESS)
+
+ def test_wait_for_downstream(self):
+ # dag_id = 'test_wait_for_downstream'
+ # dag = models.DagBag().get_dag(dag_id)
+ #
+ # TODO: clarify with airflow maintainers -- is the snippet below still true and necessary?
+ # # ti.previous_ti requires a dagrun to exist
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ # upstream, downstream = dag.tasks
+ #
+ # def run_2nd_dagrun():
+ # dag.run(
+ # start_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # end_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # executor=MockExecutor()
+ # )
+ #
+ # # doesn't run if downstream task for previous day has a null state
+ # uti_2 = TI(task=upstream, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ # uti_2.previous_ti.set_state(State.SUCCESS)
+ # self.assertEqual(uti_2.previous_ti.state, State.SUCCESS)
+ # TODO: clarify with airflow maintainers -- I don't think this triggers an AirflowException, does it?
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
Review comment:
I would like to clarify this line, because I took the original unit tests and modified them to be more comprehensive. The original author, @dima-asana , mentioned something about scheduler deadlock [here](https://github.com/apache/airflow/pull/5308/files#r285941524), which I didn't encounter. Could any kind soul enlighten me if that was the case for the older version of Airflow, but not applicable now?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] TeddyHartanto edited a comment on issue #7735:
[AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
TeddyHartanto edited a comment on issue #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#issuecomment-616093840
@ashb it's actually an old PR that has gotten all the green lights but were abandoned in the end. See [here](https://github.com/apache/airflow/pull/5308). Anyways, I think I should summon @kaxil or @potiuk since they were involved in that PR. Could you guys kindly help me check if the PR is ok? 😄 thank you!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] TeddyHartanto commented on a change in pull request
#7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
TeddyHartanto commented on a change in pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#discussion_r409399949
##########
File path: tests/models/test_taskinstance.py
##########
@@ -855,6 +855,20 @@ def test_check_task_dependencies(self, trigger_rule, successes, skipped,
self.assertEqual(completed, expect_completed)
self.assertEqual(ti.state, expect_state)
+ def test_are_dependents_done(self):
+ dag = DAG(dag_id='test_dag')
+ upstream_task = DummyOperator(dag=dag, task_id='upstream_task', start_date=DEFAULT_DATE)
+ downstream_task = DummyOperator(dag=dag, task_id='downstream_task', start_date=DEFAULT_DATE)
+ upstream_task >> downstream_task
+ uti = TI(upstream_task, DEFAULT_DATE)
+ dti = TI(downstream_task, DEFAULT_DATE)
+ for state in State.task_states:
Review comment:
OK I can change this. Initially, I was thinking of using `@parameterize.expand` too, but saw that there are a lot of task states:
```
task_states = (
SUCCESS,
RUNNING,
FAILED,
UPSTREAM_FAILED,
SKIPPED,
UP_FOR_RETRY,
UP_FOR_RESCHEDULE,
QUEUED,
NONE,
SCHEDULED,
)
``` which is why i decided to use a `for` loop that enumerates `State.task_states` for convenience.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] ashb commented on a change in pull request #7735:
[AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#discussion_r409400672
##########
File path: tests/models/test_taskinstance.py
##########
@@ -855,6 +855,20 @@ def test_check_task_dependencies(self, trigger_rule, successes, skipped,
self.assertEqual(completed, expect_completed)
self.assertEqual(ti.state, expect_state)
+ def test_are_dependents_done(self):
+ dag = DAG(dag_id='test_dag')
+ upstream_task = DummyOperator(dag=dag, task_id='upstream_task', start_date=DEFAULT_DATE)
+ downstream_task = DummyOperator(dag=dag, task_id='downstream_task', start_date=DEFAULT_DATE)
+ upstream_task >> downstream_task
+ uti = TI(upstream_task, DEFAULT_DATE)
+ dti = TI(downstream_task, DEFAULT_DATE)
+ for state in State.task_states:
Review comment:
Do you think we need to test every state, or just a few?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] TeddyHartanto commented on a change in pull request
#7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
TeddyHartanto commented on a change in pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#discussion_r409399949
##########
File path: tests/models/test_taskinstance.py
##########
@@ -855,6 +855,20 @@ def test_check_task_dependencies(self, trigger_rule, successes, skipped,
self.assertEqual(completed, expect_completed)
self.assertEqual(ti.state, expect_state)
+ def test_are_dependents_done(self):
+ dag = DAG(dag_id='test_dag')
+ upstream_task = DummyOperator(dag=dag, task_id='upstream_task', start_date=DEFAULT_DATE)
+ downstream_task = DummyOperator(dag=dag, task_id='downstream_task', start_date=DEFAULT_DATE)
+ upstream_task >> downstream_task
+ uti = TI(upstream_task, DEFAULT_DATE)
+ dti = TI(downstream_task, DEFAULT_DATE)
+ for state in State.task_states:
Review comment:
OK I can change this. Initially, I was thinking of using `@parameterize.expand` too, but saw that there are a lot of task states:
```python3
task_states = (
SUCCESS,
RUNNING,
FAILED,
UPSTREAM_FAILED,
SKIPPED,
UP_FOR_RETRY,
UP_FOR_RESCHEDULE,
QUEUED,
NONE,
SCHEDULED,
)
```
which is why i decided to use a `for` loop that enumerates `State.task_states` for convenience.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] TeddyHartanto commented on a change in pull request
#7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
TeddyHartanto commented on a change in pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#discussion_r409403720
##########
File path: tests/models/test_dagrun.py
##########
@@ -552,3 +561,131 @@ def with_all_tasks_removed(dag):
dagrun.verify_integrity()
flaky_ti.refresh_from_db()
self.assertEqual(State.NONE, flaky_ti.state)
+
+ def test_depends_on_past(self):
+ # dag_id = 'test_depends_on_past'
+ # dag = self.dagbag.get_dag(dag_id)
+ # task = dag.tasks[0]
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ # ti1 = TI(task, timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # ti2 = TI(task, timezone.datetime(2016, 1, 2, 0, 0, 0))
+ #
+ # def run_2nd_dagrun():
+ # dag.run(
+ # start_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # end_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # executor=MockExecutor()
+ # )
+ #
+ # # 2nd run of task fails by itself
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SCHEDULED)
+ #
+ # # 2nd run af task fails if 1st run of task failed
+ # ti1.set_state(State.FAILED)
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SCHEDULED)
+ #
+ # # but it works after 1st instance of task is marked as skipped
+ # ti1.set_state(State.SKIPPED)
+ # run_2nd_dagrun()
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SUCCESS)
+ #
+ # # and it also works if 1st instance is success
+ # ti2.set_state(State.NONE)
+ # ti1.set_state(State.SUCCESS)
+ # run_2nd_dagrun()
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SUCCESS)
+ dag_id = 'test_depends_on_past'
+
+ dag = self.dagbag.get_dag(dag_id)
+ task = dag.tasks[0]
+
+ self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+
+ uti1 = TI(task, timezone.datetime(2016, 1, 1, 0, 0, 0))
+ uti2 = TI(task, timezone.datetime(2016, 1, 2, 0, 0, 0))
+
+ for state in State.task_states:
+ uti1.set_state(state)
+ uti2.set_state(State.QUEUED)
+ uti2.run()
+ if state in (State.SUCCESS, State.SKIPPED):
+ self.assertEqual(uti2.state, State.SUCCESS)
+ else:
+ self.assertNotEqual(uti2.state, State.SUCCESS)
+
+ def test_wait_for_downstream(self):
+ # dag_id = 'test_wait_for_downstream'
+ # dag = models.DagBag().get_dag(dag_id)
+ #
+ # TODO: clarify with airflow maintainers -- is the snippet below still true and necessary?
+ # # ti.previous_ti requires a dagrun to exist
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ # upstream, downstream = dag.tasks
+ #
+ # def run_2nd_dagrun():
+ # dag.run(
+ # start_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # end_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # executor=MockExecutor()
+ # )
+ #
+ # # doesn't run if downstream task for previous day has a null state
+ # uti_2 = TI(task=upstream, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ # uti_2.previous_ti.set_state(State.SUCCESS)
+ # self.assertEqual(uti_2.previous_ti.state, State.SUCCESS)
+ # TODO: clarify with airflow maintainers -- I don't think this triggers an AirflowException, does it?
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
+ # uti_2.refresh_from_db()
+ # self.assertEqual(uti_2.state, State.SCHEDULED)
+ #
+ # # doesn't run if downstream task for previous day has a failed state
+ # dti_1 = TI(task=downstream, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # dti_1.set_state(State.FAILED)
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
+ # uti_2.refresh_from_db()
+ # self.assertEqual(uti_2.state, State.SCHEDULED)
+ #
+ # # runs if downstream task for previous day has a skipped state
+ # dti_1 = TI(task=downstream, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # dti_1.set_state(State.SKIPPED)
+ # run_2nd_dagrun()
+ # uti_2.refresh_from_db()
+ # self.assertEqual(uti_2.state, State.SUCCESS)
+ #
+ # # runs if downstream task for previous day has a success state
+ # uti_2.set_state(State.NONE)
+ # dti_1.set_state(State.SUCCESS)
+ # run_2nd_dagrun()
+ # uti_2.refresh_from_db()
+ # self.assertEqual(uti_2.state, State.SUCCESS)
+ dag_id = 'test_wait_for_downstream'
+ dag = self.dagbag.get_dag(dag_id)
+ upstream, downstream = dag.tasks
+
+ # For ti.set_state() to work, the DagRun has to exist,
+ # Otherwise ti.previous_ti returns an unpersisted TI
+ self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+
+ dti1 = TI(task=downstream, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ uti2 = TI(task=upstream, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ uti2.previous_ti.set_state(State.SUCCESS)
+ self.assertEqual(uti2.previous_ti.state, State.SUCCESS)
+
+ for state in State.task_states:
+ dti1.set_state(state)
+ uti2.set_state(State.QUEUED)
+ uti2.run()
+ if state in (State.SUCCESS, State.SKIPPED):
+ self.assertEqual(uti2.state, State.SUCCESS)
+ else:
+ self.assertNotEqual(uti2.state, State.SUCCESS)
Review comment:
The one in `test_taskinstance.py` is really like a pure unit test. There is no `wait_for_downstream` flag being used anywhere there. It's really just a test on the behavior of `TaskInstance.are_dependents_done()`.
Meanwhile, the tests in `test_dagrun.py` has a wider scope in which the `wait_for_downstream=True` or `depends_on_past=True` are set when initialising the `Operator`'s. The state of the `previous_ti_downstream` and `previous_ti` are manipulated, `ti` is ran, and then we assert what we expect should be the state of `ti`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] TeddyHartanto opened a new pull request #7735:
[AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
TeddyHartanto opened a new pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735
Before this fix, tasks that are in SUCCESS or SKIPPED state
satisfy the depends_on_past check, but only tasks that are in
SUCCESS state satisfy the wait_for_downstream check. That makes the API
inconsistent and unintuitive.
---
Issue link: [AIRFLOW-4549](https://issues.apache.org/jira/browse/AIRFLOW-4549)
Make sure to mark the boxes below before creating PR: [x]
- [x] Description above provides context of the change
- [x] Commit message/PR title starts with `[AIRFLOW-NNNN]`. AIRFLOW-NNNN = JIRA ID<sup>*</sup>
- [x] Unit tests coverage for changes (not needed for documentation changes)
- [x] Commits follow "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)"
- [x] Relevant documentation is updated including usage instructions.
- [x] I will engage committers as explained in [Contribution Workflow Example](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#contribution-workflow-example).
<sup>*</sup> For document-only changes commit message can start with `[AIRFLOW-XXXX]`.
---
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/master/UPDATING.md).
Read the [Pull Request Guidelines](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#pull-request-guidelines) for more information.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] TeddyHartanto commented on a change in pull request
#7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
TeddyHartanto commented on a change in pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#discussion_r409403720
##########
File path: tests/models/test_dagrun.py
##########
@@ -552,3 +561,131 @@ def with_all_tasks_removed(dag):
dagrun.verify_integrity()
flaky_ti.refresh_from_db()
self.assertEqual(State.NONE, flaky_ti.state)
+
+ def test_depends_on_past(self):
+ # dag_id = 'test_depends_on_past'
+ # dag = self.dagbag.get_dag(dag_id)
+ # task = dag.tasks[0]
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ # ti1 = TI(task, timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # ti2 = TI(task, timezone.datetime(2016, 1, 2, 0, 0, 0))
+ #
+ # def run_2nd_dagrun():
+ # dag.run(
+ # start_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # end_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # executor=MockExecutor()
+ # )
+ #
+ # # 2nd run of task fails by itself
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SCHEDULED)
+ #
+ # # 2nd run af task fails if 1st run of task failed
+ # ti1.set_state(State.FAILED)
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SCHEDULED)
+ #
+ # # but it works after 1st instance of task is marked as skipped
+ # ti1.set_state(State.SKIPPED)
+ # run_2nd_dagrun()
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SUCCESS)
+ #
+ # # and it also works if 1st instance is success
+ # ti2.set_state(State.NONE)
+ # ti1.set_state(State.SUCCESS)
+ # run_2nd_dagrun()
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SUCCESS)
+ dag_id = 'test_depends_on_past'
+
+ dag = self.dagbag.get_dag(dag_id)
+ task = dag.tasks[0]
+
+ self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+
+ uti1 = TI(task, timezone.datetime(2016, 1, 1, 0, 0, 0))
+ uti2 = TI(task, timezone.datetime(2016, 1, 2, 0, 0, 0))
+
+ for state in State.task_states:
+ uti1.set_state(state)
+ uti2.set_state(State.QUEUED)
+ uti2.run()
+ if state in (State.SUCCESS, State.SKIPPED):
+ self.assertEqual(uti2.state, State.SUCCESS)
+ else:
+ self.assertNotEqual(uti2.state, State.SUCCESS)
+
+ def test_wait_for_downstream(self):
+ # dag_id = 'test_wait_for_downstream'
+ # dag = models.DagBag().get_dag(dag_id)
+ #
+ # TODO: clarify with airflow maintainers -- is the snippet below still true and necessary?
+ # # ti.previous_ti requires a dagrun to exist
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ # upstream, downstream = dag.tasks
+ #
+ # def run_2nd_dagrun():
+ # dag.run(
+ # start_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # end_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # executor=MockExecutor()
+ # )
+ #
+ # # doesn't run if downstream task for previous day has a null state
+ # uti_2 = TI(task=upstream, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ # uti_2.previous_ti.set_state(State.SUCCESS)
+ # self.assertEqual(uti_2.previous_ti.state, State.SUCCESS)
+ # TODO: clarify with airflow maintainers -- I don't think this triggers an AirflowException, does it?
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
+ # uti_2.refresh_from_db()
+ # self.assertEqual(uti_2.state, State.SCHEDULED)
+ #
+ # # doesn't run if downstream task for previous day has a failed state
+ # dti_1 = TI(task=downstream, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # dti_1.set_state(State.FAILED)
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
+ # uti_2.refresh_from_db()
+ # self.assertEqual(uti_2.state, State.SCHEDULED)
+ #
+ # # runs if downstream task for previous day has a skipped state
+ # dti_1 = TI(task=downstream, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # dti_1.set_state(State.SKIPPED)
+ # run_2nd_dagrun()
+ # uti_2.refresh_from_db()
+ # self.assertEqual(uti_2.state, State.SUCCESS)
+ #
+ # # runs if downstream task for previous day has a success state
+ # uti_2.set_state(State.NONE)
+ # dti_1.set_state(State.SUCCESS)
+ # run_2nd_dagrun()
+ # uti_2.refresh_from_db()
+ # self.assertEqual(uti_2.state, State.SUCCESS)
+ dag_id = 'test_wait_for_downstream'
+ dag = self.dagbag.get_dag(dag_id)
+ upstream, downstream = dag.tasks
+
+ # For ti.set_state() to work, the DagRun has to exist,
+ # Otherwise ti.previous_ti returns an unpersisted TI
+ self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+
+ dti1 = TI(task=downstream, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ uti2 = TI(task=upstream, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ uti2.previous_ti.set_state(State.SUCCESS)
+ self.assertEqual(uti2.previous_ti.state, State.SUCCESS)
+
+ for state in State.task_states:
+ dti1.set_state(state)
+ uti2.set_state(State.QUEUED)
+ uti2.run()
+ if state in (State.SUCCESS, State.SKIPPED):
+ self.assertEqual(uti2.state, State.SUCCESS)
+ else:
+ self.assertNotEqual(uti2.state, State.SUCCESS)
Review comment:
The one in `test_taskinstance.py` is really like a pure unit test. There is no `wait_for_downstream` flag being used anywhere there. It's really just a test on the behavior of `TaskInstance.are_dependents_done()`.
Meanwhile, the tests in `test_dagrun.py` has a wider scope in which the `wait_for_downstream=True` or `depends_on_past=True` are set when initialising the `Operator`'s. The state of the `previous_ti_downstream` and `previous_ti` are manipulated, `ti` is ran, and then we assert what we expect the state of `ti` should be.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] ashb commented on a change in pull request #7735:
[AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#discussion_r409396318
##########
File path: tests/models/test_taskinstance.py
##########
@@ -855,6 +855,20 @@ def test_check_task_dependencies(self, trigger_rule, successes, skipped,
self.assertEqual(completed, expect_completed)
self.assertEqual(ti.state, expect_state)
+ def test_are_dependents_done(self):
+ dag = DAG(dag_id='test_dag')
+ upstream_task = DummyOperator(dag=dag, task_id='upstream_task', start_date=DEFAULT_DATE)
+ downstream_task = DummyOperator(dag=dag, task_id='downstream_task', start_date=DEFAULT_DATE)
+ upstream_task >> downstream_task
+ uti = TI(upstream_task, DEFAULT_DATE)
+ dti = TI(downstream_task, DEFAULT_DATE)
+ for state in State.task_states:
Review comment:
This test is a bit hard to work out what it is doing, and should instead be refactored to use `@parameterize.expand`, something like this:
```python
@parameterize.expand(
(State.SUCCESS, True),
(State.SKIPPED, True),
(State.RUNNING, True),
)
def test_are_dependents_done(self, current_state, expected_outcome):
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] HaloKo4 commented on issue #7735: [AIRFLOW-4549] Allow
skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
HaloKo4 commented on issue #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#issuecomment-609638482
I think there is an issue with the test
```
def get_dag(subdir: Optional[str], dag_id: str) -> DAG:
"""Returns DAG of a given dag_id"""
dagbag = DagBag(process_subdir(subdir))
if dag_id not in dagbag.dags:
raise AirflowException(
'dag_id could not be found: {}. Either the dag did not exist or it failed to '
> 'parse.'.format(dag_id))
E airflow.exceptions.AirflowException: dag_id could not be found: test_depends_on_past. Either the dag did not exist or it failed to parse.
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] TeddyHartanto commented on a change in pull request
#7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
TeddyHartanto commented on a change in pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#discussion_r409389154
##########
File path: tests/models/test_dagrun.py
##########
@@ -552,3 +561,131 @@ def with_all_tasks_removed(dag):
dagrun.verify_integrity()
flaky_ti.refresh_from_db()
self.assertEqual(State.NONE, flaky_ti.state)
+
+ def test_depends_on_past(self):
+ # dag_id = 'test_depends_on_past'
Review comment:
I preserved the tests written by the original author, @dima-asana for the benefit of reference to the Airflow contributors. After getting clarification on this, I will delete the comments in the tests before we proceed to merge the PR
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] TeddyHartanto commented on a change in pull request
#7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
TeddyHartanto commented on a change in pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#discussion_r409367097
##########
File path: tests/models/test_dagrun.py
##########
@@ -552,3 +561,131 @@ def with_all_tasks_removed(dag):
dagrun.verify_integrity()
flaky_ti.refresh_from_db()
self.assertEqual(State.NONE, flaky_ti.state)
+
+ def test_depends_on_past(self):
+ # dag_id = 'test_depends_on_past'
+ # dag = self.dagbag.get_dag(dag_id)
+ # task = dag.tasks[0]
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ # ti1 = TI(task, timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # ti2 = TI(task, timezone.datetime(2016, 1, 2, 0, 0, 0))
+ #
+ # def run_2nd_dagrun():
+ # dag.run(
+ # start_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # end_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # executor=MockExecutor()
+ # )
+ #
+ # # 2nd run of task fails by itself
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SCHEDULED)
+ #
+ # # 2nd run af task fails if 1st run of task failed
+ # ti1.set_state(State.FAILED)
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SCHEDULED)
+ #
+ # # but it works after 1st instance of task is marked as skipped
+ # ti1.set_state(State.SKIPPED)
+ # run_2nd_dagrun()
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SUCCESS)
+ #
+ # # and it also works if 1st instance is success
+ # ti2.set_state(State.NONE)
+ # ti1.set_state(State.SUCCESS)
+ # run_2nd_dagrun()
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SUCCESS)
+ dag_id = 'test_depends_on_past'
+
+ dag = self.dagbag.get_dag(dag_id)
+ task = dag.tasks[0]
+
+ self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+
+ uti1 = TI(task, timezone.datetime(2016, 1, 1, 0, 0, 0))
+ uti2 = TI(task, timezone.datetime(2016, 1, 2, 0, 0, 0))
+
+ for state in State.task_states:
+ uti1.set_state(state)
+ uti2.set_state(State.QUEUED)
+ uti2.run()
+ if state in (State.SUCCESS, State.SKIPPED):
+ self.assertEqual(uti2.state, State.SUCCESS)
+ else:
+ self.assertNotEqual(uti2.state, State.SUCCESS)
+
+ def test_wait_for_downstream(self):
+ # dag_id = 'test_wait_for_downstream'
+ # dag = models.DagBag().get_dag(dag_id)
+ #
+ # TODO: clarify with airflow maintainers -- is the snippet below still true and necessary?
+ # # ti.previous_ti requires a dagrun to exist
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ # upstream, downstream = dag.tasks
+ #
+ # def run_2nd_dagrun():
+ # dag.run(
+ # start_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # end_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # executor=MockExecutor()
+ # )
+ #
+ # # doesn't run if downstream task for previous day has a null state
+ # uti_2 = TI(task=upstream, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ # uti_2.previous_ti.set_state(State.SUCCESS)
+ # self.assertEqual(uti_2.previous_ti.state, State.SUCCESS)
+ # TODO: clarify with airflow maintainers -- I don't think this triggers an AirflowException, does it?
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
Review comment:
I would like to clarify this line, because I took the original unit tests and modified them to be more comprehensive. The original author, @dima-asana , mentioned something about scheduler deadlock [here](https://github.com/apache/airflow/pull/5308/files#r285941524), which I didn't encounter. Could any kind soul enlighten me if that was the case for the older version of Airflow, but not applicable now? Or did I miss something?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] TeddyHartanto edited a comment on issue #7735:
[AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
TeddyHartanto edited a comment on issue #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#issuecomment-616093840
@ashb it's actually an old PR that has gotten all the green lights but were abandoned in the end. See [here](https://github.com/apache/airflow/pull/5308). Anyways, no worry. Let me summon @kaxil or @potiuk since they were involved in that PR. Could you guys kindly help me check if the PR is ok? 😬 thank you!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] TeddyHartanto commented on a change in pull request
#7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
TeddyHartanto commented on a change in pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#discussion_r409399949
##########
File path: tests/models/test_taskinstance.py
##########
@@ -855,6 +855,20 @@ def test_check_task_dependencies(self, trigger_rule, successes, skipped,
self.assertEqual(completed, expect_completed)
self.assertEqual(ti.state, expect_state)
+ def test_are_dependents_done(self):
+ dag = DAG(dag_id='test_dag')
+ upstream_task = DummyOperator(dag=dag, task_id='upstream_task', start_date=DEFAULT_DATE)
+ downstream_task = DummyOperator(dag=dag, task_id='downstream_task', start_date=DEFAULT_DATE)
+ upstream_task >> downstream_task
+ uti = TI(upstream_task, DEFAULT_DATE)
+ dti = TI(downstream_task, DEFAULT_DATE)
+ for state in State.task_states:
Review comment:
OK I can change this. Initially, I was thinking of using `@parameterize.expand` too, but saw that there are a lot of task states:
```python3
task_states = (
SUCCESS,
RUNNING,
FAILED,
UPSTREAM_FAILED,
SKIPPED,
UP_FOR_RETRY,
UP_FOR_RESCHEDULE,
QUEUED,
NONE,
SCHEDULED,
)
``` which is why i decided to use a `for` loop that enumerates `State.task_states` for convenience.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] ashb commented on a change in pull request #7735:
[AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#discussion_r409396318
##########
File path: tests/models/test_taskinstance.py
##########
@@ -855,6 +855,20 @@ def test_check_task_dependencies(self, trigger_rule, successes, skipped,
self.assertEqual(completed, expect_completed)
self.assertEqual(ti.state, expect_state)
+ def test_are_dependents_done(self):
+ dag = DAG(dag_id='test_dag')
+ upstream_task = DummyOperator(dag=dag, task_id='upstream_task', start_date=DEFAULT_DATE)
+ downstream_task = DummyOperator(dag=dag, task_id='downstream_task', start_date=DEFAULT_DATE)
+ upstream_task >> downstream_task
+ uti = TI(upstream_task, DEFAULT_DATE)
+ dti = TI(downstream_task, DEFAULT_DATE)
+ for state in State.task_states:
Review comment:
This test is a bit hard to work out what it is doing, and should instead be refactored to use `@parameterize.expand`, something like this:
```python
@parameterize.expand(
(State.SUCCESS, True),
(State.SKIPPED, True),
(State.RUNNING, False),
)
def test_are_dependents_done(self, current_state, expected_outcome):
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] TeddyHartanto commented on a change in pull request
#7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
TeddyHartanto commented on a change in pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#discussion_r409367097
##########
File path: tests/models/test_dagrun.py
##########
@@ -552,3 +561,131 @@ def with_all_tasks_removed(dag):
dagrun.verify_integrity()
flaky_ti.refresh_from_db()
self.assertEqual(State.NONE, flaky_ti.state)
+
+ def test_depends_on_past(self):
+ # dag_id = 'test_depends_on_past'
+ # dag = self.dagbag.get_dag(dag_id)
+ # task = dag.tasks[0]
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ # ti1 = TI(task, timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # ti2 = TI(task, timezone.datetime(2016, 1, 2, 0, 0, 0))
+ #
+ # def run_2nd_dagrun():
+ # dag.run(
+ # start_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # end_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # executor=MockExecutor()
+ # )
+ #
+ # # 2nd run of task fails by itself
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SCHEDULED)
+ #
+ # # 2nd run af task fails if 1st run of task failed
+ # ti1.set_state(State.FAILED)
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SCHEDULED)
+ #
+ # # but it works after 1st instance of task is marked as skipped
+ # ti1.set_state(State.SKIPPED)
+ # run_2nd_dagrun()
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SUCCESS)
+ #
+ # # and it also works if 1st instance is success
+ # ti2.set_state(State.NONE)
+ # ti1.set_state(State.SUCCESS)
+ # run_2nd_dagrun()
+ # ti2.refresh_from_db()
+ # self.assertEqual(ti2.state, State.SUCCESS)
+ dag_id = 'test_depends_on_past'
+
+ dag = self.dagbag.get_dag(dag_id)
+ task = dag.tasks[0]
+
+ self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+
+ uti1 = TI(task, timezone.datetime(2016, 1, 1, 0, 0, 0))
+ uti2 = TI(task, timezone.datetime(2016, 1, 2, 0, 0, 0))
+
+ for state in State.task_states:
+ uti1.set_state(state)
+ uti2.set_state(State.QUEUED)
+ uti2.run()
+ if state in (State.SUCCESS, State.SKIPPED):
+ self.assertEqual(uti2.state, State.SUCCESS)
+ else:
+ self.assertNotEqual(uti2.state, State.SUCCESS)
+
+ def test_wait_for_downstream(self):
+ # dag_id = 'test_wait_for_downstream'
+ # dag = models.DagBag().get_dag(dag_id)
+ #
+ # TODO: clarify with airflow maintainers -- is the snippet below still true and necessary?
+ # # ti.previous_ti requires a dagrun to exist
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0))
+ # self.create_dag_run(dag, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ # upstream, downstream = dag.tasks
+ #
+ # def run_2nd_dagrun():
+ # dag.run(
+ # start_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # end_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
+ # executor=MockExecutor()
+ # )
+ #
+ # # doesn't run if downstream task for previous day has a null state
+ # uti_2 = TI(task=upstream, execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0))
+ # uti_2.previous_ti.set_state(State.SUCCESS)
+ # self.assertEqual(uti_2.previous_ti.state, State.SUCCESS)
+ # TODO: clarify with airflow maintainers -- I don't think this triggers an AirflowException, does it?
+ # self.assertRaises(AirflowException, run_2nd_dagrun)
Review comment:
I would like to clarify this line, because I took the original unit tests and modified them to be more comprehensive. The original author, @dima-asana , mentioned something about scheduler deadlock [here](https://github.com/apache/airflow/pull/5308/files#r285941524), which I didn't encounter. Could any kind soul enlighten me if that was the case for the older version of Airflow, but not the case anymore?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] TeddyHartanto commented on issue #7735: [AIRFLOW-4549]
Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
TeddyHartanto commented on issue #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#issuecomment-614993827
Hi @ashb, I've fixed the tests :) please let me know if there's anything else that I miss
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] TeddyHartanto closed pull request #7735: [AIRFLOW-4549]
Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
TeddyHartanto closed pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] TeddyHartanto commented on a change in pull request
#7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
TeddyHartanto commented on a change in pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#discussion_r409389154
##########
File path: tests/models/test_dagrun.py
##########
@@ -552,3 +561,131 @@ def with_all_tasks_removed(dag):
dagrun.verify_integrity()
flaky_ti.refresh_from_db()
self.assertEqual(State.NONE, flaky_ti.state)
+
+ def test_depends_on_past(self):
+ # dag_id = 'test_depends_on_past'
Review comment:
After getting clarification on this, I will delete the comments in the tests
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] TeddyHartanto commented on a change in pull request
#7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
TeddyHartanto commented on a change in pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#discussion_r409399949
##########
File path: tests/models/test_taskinstance.py
##########
@@ -855,6 +855,20 @@ def test_check_task_dependencies(self, trigger_rule, successes, skipped,
self.assertEqual(completed, expect_completed)
self.assertEqual(ti.state, expect_state)
+ def test_are_dependents_done(self):
+ dag = DAG(dag_id='test_dag')
+ upstream_task = DummyOperator(dag=dag, task_id='upstream_task', start_date=DEFAULT_DATE)
+ downstream_task = DummyOperator(dag=dag, task_id='downstream_task', start_date=DEFAULT_DATE)
+ upstream_task >> downstream_task
+ uti = TI(upstream_task, DEFAULT_DATE)
+ dti = TI(downstream_task, DEFAULT_DATE)
+ for state in State.task_states:
Review comment:
OK I can change this. Initially, I was thinking of using `@parameterize.expand` too, but saw that there are a lot of task states:
```python3
task_states = (
SUCCESS,
RUNNING,
FAILED,
UPSTREAM_FAILED,
SKIPPED,
UP_FOR_RETRY,
UP_FOR_RESCHEDULE,
QUEUED,
NONE,
SCHEDULED,
)
```
which is why i decided to use a `for` loop that enumerates `State.task_states` for convenience.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] ashb commented on issue #7735: [AIRFLOW-4549] Allow
skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
ashb commented on issue #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#issuecomment-615841790
Code looks fine, just need other people to weigh in on if this makes sense as a change - I just don't have head space to work that out right now, sorry
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] TeddyHartanto commented on a change in pull request
#7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
TeddyHartanto commented on a change in pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#discussion_r409405257
##########
File path: tests/models/test_taskinstance.py
##########
@@ -855,6 +855,20 @@ def test_check_task_dependencies(self, trigger_rule, successes, skipped,
self.assertEqual(completed, expect_completed)
self.assertEqual(ti.state, expect_state)
+ def test_are_dependents_done(self):
+ dag = DAG(dag_id='test_dag')
+ upstream_task = DummyOperator(dag=dag, task_id='upstream_task', start_date=DEFAULT_DATE)
+ downstream_task = DummyOperator(dag=dag, task_id='downstream_task', start_date=DEFAULT_DATE)
+ upstream_task >> downstream_task
+ uti = TI(upstream_task, DEFAULT_DATE)
+ dti = TI(downstream_task, DEFAULT_DATE)
+ for state in State.task_states:
Review comment:
Just a few is sufficient actually.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] ashb commented on a change in pull request #7735:
[AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#discussion_r409396318
##########
File path: tests/models/test_taskinstance.py
##########
@@ -855,6 +855,20 @@ def test_check_task_dependencies(self, trigger_rule, successes, skipped,
self.assertEqual(completed, expect_completed)
self.assertEqual(ti.state, expect_state)
+ def test_are_dependents_done(self):
+ dag = DAG(dag_id='test_dag')
+ upstream_task = DummyOperator(dag=dag, task_id='upstream_task', start_date=DEFAULT_DATE)
+ downstream_task = DummyOperator(dag=dag, task_id='downstream_task', start_date=DEFAULT_DATE)
+ upstream_task >> downstream_task
+ uti = TI(upstream_task, DEFAULT_DATE)
+ dti = TI(downstream_task, DEFAULT_DATE)
+ for state in State.task_states:
Review comment:
This test is a bit hard to work out what it is doing, and should instead be refactored to use `@parameterize.expand`, something like this:
```python
@parameterize.expand(
(State.SUCCESS, True),
(State.SKIPPED, True),
(State.RUNNING, True),
def test_are_dependents_done(self, current_state, expected_outcome):
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [airflow] TeddyHartanto commented on issue #7735: [AIRFLOW-4549]
Allow skipped tasks to satisfy wait_for_downstream
Posted by GitBox <gi...@apache.org>.
TeddyHartanto commented on issue #7735: [AIRFLOW-4549] Allow skipped tasks to satisfy wait_for_downstream
URL: https://github.com/apache/airflow/pull/7735#issuecomment-616093840
@ashb it's actually an old PR that has gotten all the green lights but were abandoned in the end. See [here](https://github.com/apache/airflow/pull/5308). Anyways, I think I should summon @kaxil or @potiuk. Could you guys kindly help me check if the PR is ok? 😄 thank you!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services