Resolve upstream tasks when template field is XComArg #8805

turbaszek · 2020-05-10T11:19:26Z

closes: #8054

Make sure to mark the boxes below before creating PR: [x]

Description above provides context of the change
Unit tests coverage for changes (not needed for documentation changes)
Target Github ISSUE in description if exists
Commits follow "How to write a good git commit message"
Relevant documentation is updated including usage instructions.
I will engage committers as explained in Contribution Workflow Example.

In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.
Read the Pull Request Guidelines for more information.

airflow/models/baseoperator.py

tests/models/test_baseoperator.py

evgenyshulman · 2020-05-10T20:11:28Z

airflow/models/baseoperator.py

-        for field in self.template_fields:
-            arg = getattr(self, field)
+
+        def apply_set_upstream(arg: Any):
            if isinstance(arg, XComArg):


this traversing part should be very "error prune". wee are starting to traverse all user templated args with no exception (old code is affected), it can be that something will fail and this feature will become a "blocker" to update to the new version

it can be that something will fail and this feature will become a "blocker" to update to the new version

Can you provide an example? We perform an action only when we encounter XComArg so self.set_upstream should work. If it doesn't work then it means that our implementation is corrupted.

airflow/models/baseoperator.py

airflow/example_dags/example_xcomargs.py

airflow/models/baseoperator.py

airflow/models/dag.py

airflow/providers/google/cloud/operators/automl.py

airflow/providers/google/marketing_platform/operators/display_video.py

airflow/models/baseoperator.py

ashb

Please add a test that checks the up/down stream dependencies work with Dag Serialization enabled.

turbaszek · 2020-05-19T10:20:08Z

@ash @kaxil @evgenyshulman are we still sure that this way it's better than having a metaclass?

ashb

templated classes aren't handled yet.

Additionally, https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-31%3A+Airflow+functional+DAG+definition doesn't have this .output attribute.

Where has this concept come from? It wasn't in the AIP.

airflow/models/baseoperator.py

turbaszek · 2020-05-19T13:20:42Z

Where has this concept come from? It wasn't in the AIP.

It was proposed in #8055 . Without it mixing functional and non-functional operators will be hard.

turbaszek · 2020-05-26T08:48:18Z

@ashb @evgenyshulman @casassg I think all questions were addressed

casassg

lgtm overall 🚀

tests/models/test_baseoperator.py

closes: apache#8054

airflow/models/baseoperator.py

tests/test_utils/__init__.py

Co-authored-by: Ash Berlin-Taylor <ash_github@firemirror.com>

airflow/models/baseoperator.py

* Resolve upstream tasks when template field is XComArg closes: apache#8054 * fixup! Resolve upstream tasks when template field is XComArg * Resolve task relations in DagRun and DagBag * Add tests for serialized DAG * Set dependencies only in bag_dag, refactor tests * Traverse template_fields attribute * Use provide_test_dag_bag in all tests * fixup! Use provide_test_dag_bag in all tests * Use metaclass + setattr * Add prepare_for_execution method * Check signature of __init__ not class * Apply suggestions from code review Co-authored-by: Ash Berlin-Taylor <ash_github@firemirror.com> * Update airflow/models/baseoperator.py Co-authored-by: Ash Berlin-Taylor <ash_github@firemirror.com>

When we are referring ``{{task}}` in jinja templates we can also refer some of the fields, which are templated. We are not able to solve all the problems with such rendering (specifically recursive rendering of the fields used in JINJA templating might be problematic. Currently whether you see original, or rendered field depends solely on the sequence in templated_fields. However that would not even explain the rendering problem described in apache#13559 where kwargs were defined after opargs and the rendering of opargs **should** work. It turned out that the problem was with a change introduced in apache#8805 which made the context effectively holds a DIFFERENT task than the current one. Context held an original task, and the curren task was actually a locked copy of it (to allow resolving upstream args before locking). As a result, any changes done by rendering templates were not visible in the task accessed via {{ task }} jinja variable. This change replaces the the task stored in context with the same copy that is then used later during execution so that at least the "sequential" rendering works and templated fields which are 'earlier' in the list of templated fields can be used (and render correctly) in the following fields. Fixes: apache#13559

When we are referring ``{{task}}` in jinja templates we can also refer some of the fields, which are templated. We are not able to solve all the problems with such rendering (specifically recursive rendering of the fields used in JINJA templating might be problematic. Currently whether you see original, or rendered field depends solely on the sequence in templated_fields. However that would not even explain the rendering problem described in #13559 where kwargs were defined after opargs and the rendering of opargs **should** work. It turned out that the problem was with a change introduced in #8805 which made the context effectively holds a DIFFERENT task than the current one. Context held an original task, and the curren task was actually a locked copy of it (to allow resolving upstream args before locking). As a result, any changes done by rendering templates were not visible in the task accessed via {{ task }} jinja variable. This change replaces the the task stored in context with the same copy that is then used later during execution so that at least the "sequential" rendering works and templated fields which are 'earlier' in the list of templated fields can be used (and render correctly) in the following fields. Fixes: #13559

turbaszek requested a review from kaxil May 10, 2020 11:19

turbaszek mentioned this pull request May 10, 2020

[AIP-31] Implement XComArg model to functionally pass output from one operator to the next #8652

Merged

6 tasks

turbaszek commented May 10, 2020

View reviewed changes

airflow/models/baseoperator.py Show resolved Hide resolved

turbaszek requested review from dimberman and mik-laj May 10, 2020 11:22

evgenyshulman reviewed May 10, 2020

View reviewed changes

airflow/models/baseoperator.py Outdated Show resolved Hide resolved

airflow/models/baseoperator.py Show resolved Hide resolved

tests/models/test_baseoperator.py Outdated Show resolved Hide resolved

turbaszek force-pushed the aip31-set-relations branch from d40c9d9 to 252be10 Compare May 10, 2020 13:49

evgenyshulman reviewed May 10, 2020

View reviewed changes

turbaszek requested a review from ashb May 11, 2020 11:48

turbaszek added the AIP-31 Task Flow API for nicer DAG definition label May 11, 2020

ashb reviewed May 11, 2020

View reviewed changes

airflow/models/baseoperator.py Show resolved Hide resolved

turbaszek requested a review from ashb May 12, 2020 09:57

ashb reviewed May 14, 2020

View reviewed changes

airflow/example_dags/example_xcomargs.py Outdated Show resolved Hide resolved

ashb reviewed May 14, 2020

View reviewed changes

airflow/models/baseoperator.py Show resolved Hide resolved

ashb reviewed May 14, 2020

View reviewed changes

airflow/models/dag.py Outdated Show resolved Hide resolved

ashb reviewed May 14, 2020

View reviewed changes

airflow/providers/google/cloud/operators/automl.py Outdated Show resolved Hide resolved

ashb reviewed May 14, 2020

View reviewed changes

airflow/providers/google/marketing_platform/operators/display_video.py Outdated Show resolved Hide resolved

ashb reviewed May 14, 2020

View reviewed changes

airflow/models/baseoperator.py Show resolved Hide resolved

ashb requested changes May 14, 2020

View reviewed changes

turbaszek force-pushed the aip31-set-relations branch from 0660ed3 to 4b4af57 Compare May 14, 2020 12:36

turbaszek requested a review from ashb May 19, 2020 12:13

ashb requested changes May 19, 2020

View reviewed changes

airflow/models/baseoperator.py Show resolved Hide resolved

turbaszek force-pushed the aip31-set-relations branch 3 times, most recently from 46eab09 to 6743e4e Compare May 21, 2020 10:00

turbaszek requested a review from ashb May 21, 2020 12:51

casassg approved these changes May 27, 2020

View reviewed changes

tests/models/test_baseoperator.py Outdated Show resolved Hide resolved

turbaszek added 11 commits June 9, 2020 19:26

Resolve upstream tasks when template field is XComArg

daf4299

closes: apache#8054

fixup! Resolve upstream tasks when template field is XComArg

000c327

Resolve task relations in DagRun and DagBag

0f1f029

Add tests for serialized DAG

bd91914

Set dependencies only in bag_dag, refactor tests

f29a191

Traverse template_fields attribute

e3115ff

Use provide_test_dag_bag in all tests

75872df

fixup! Use provide_test_dag_bag in all tests

ff5c4c0

Use metaclass + setattr

a4944bd

Add prepare_for_execution method

60e87b9

Check signature of __init__ not class

745d0a3

turbaszek force-pushed the aip31-set-relations branch from 3536855 to 745d0a3 Compare June 9, 2020 17:27

ashb reviewed Jun 15, 2020

View reviewed changes

airflow/models/baseoperator.py Outdated Show resolved Hide resolved

ashb reviewed Jun 15, 2020

View reviewed changes

tests/test_utils/__init__.py Outdated Show resolved Hide resolved

Apply suggestions from code review

107a70a

Co-authored-by: Ash Berlin-Taylor <ash_github@firemirror.com>

ashb approved these changes Jun 15, 2020

View reviewed changes

turbaszek commented Jun 15, 2020

View reviewed changes

airflow/models/baseoperator.py Outdated Show resolved Hide resolved

Update airflow/models/baseoperator.py

cfc9ab5

turbaszek merged commit 431ea32 into apache:master Jun 15, 2020

turbaszek mentioned this pull request Jun 18, 2020

Add info about BaseOperatorMeta to UPDATING.md #9369

Merged

6 tasks

potiuk mentioned this pull request Sep 25, 2021

Fix accessing rendered task.x attributes from within templates #18516

Merged

potiuk mentioned this pull request Sep 25, 2021

Nested templated variables do not always render #13559

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Resolve upstream tasks when template field is XComArg #8805

Resolve upstream tasks when template field is XComArg #8805

turbaszek commented May 10, 2020

evgenyshulman May 10, 2020

turbaszek May 11, 2020

ashb left a comment

turbaszek commented May 19, 2020

ashb left a comment

turbaszek commented May 19, 2020

turbaszek commented May 26, 2020

casassg left a comment

Resolve upstream tasks when template field is XComArg #8805

Resolve upstream tasks when template field is XComArg #8805

Conversation

turbaszek commented May 10, 2020

evgenyshulman May 10, 2020

Choose a reason for hiding this comment

turbaszek May 11, 2020

Choose a reason for hiding this comment

ashb left a comment

Choose a reason for hiding this comment

turbaszek commented May 19, 2020

ashb left a comment

Choose a reason for hiding this comment

turbaszek commented May 19, 2020

turbaszek commented May 26, 2020

casassg left a comment

Choose a reason for hiding this comment