[AIRFLOW-198] Implement 'only_run_latest' feature in BaseOperator

[PeterAttardo]

Dear Airflow Maintainers,

Please accept this PR that addresses the following issues:

• https://issues.apache.org/jira/browse/AIRFLOW-198

[criccomini]

Having an is_ method change state is a little unintuitive. @bolkedebruin just did some scheduler work to clean up this kind of pattern. Can you split the is_ method from the set call, so that change state is explicit, rather than done via what appears to be a read method?

[PeterAttardo]

Agreed that it's not the cleanest, but I'm not sure how easily it can be extracted. It behaves very similarly to evaluate_trigger_rule() which gets called from are_dependencies_met() in the same chain. Is there a work in progress branch anywhere that shows how state assignments in evaluate_trigger_rule() were brought out of the is_ stack that I could reference? The other change I could make would be to add a flag similar to flag_upstream_failed so that state would only change if the original caller explicitly passed through the flag enabling it.

[mistercrunch]

you can also just change the name of the method to reflect what is taking place

[criccomini]

One thing that I notice about this PR is that it doesn't appear to try to execute the most recent execution_date first in a case where there are multiple DAG runs that need to be run. For example, if you have 10 DAG runs that all need to be run, and only_run_latest=True, then it makes sense to run the most recent one first and skip the other 9.

@bolkedebruin @jlowin @mistercrunch Do we want to introduce a new state for this (FUTURE_SUCCEEDED) or just use SKIPPED?

This PR should include some tests as well.

[PeterAttardo]

I considered the issue to be two related issues:

1. For any task B, which depends on task A (marked as only_run_latest), the dependencies for an instance of B should be met if any instance of A has succeeded on or after the execution date.
2. Avoid scheduling multiple instances of a task that has been marked as only_run_latest and prioritize the most recent execution date.

This PR only addresses the first of those. The second piece raised many more architectural questions and potential changes to the way Airflow schedules jobs. I didn't want to commit to a potentially large architectural change before seeing how the maintainers were thinking about designing for this feature.

[mistercrunch]

TaskInstance.state.in_((State.SUCCESS, State.FUTURE_SUCCEEDED))

[mistercrunch]

Before I go further in the review, I think this may conflict with #1525. @aoen?

[aoen]

Yes it conflicts, thanks for finding this Max. Under the new model (you can see in my PR that max linked #1525 ) you would create a dependency class for future succeeded.

[PeterAttardo]

#1525 looks like it would allow this functionality to slot in nicely. I can circle back to this issue once #1525 has been merged and approach it in the new idiomatic way.

[r39132]

This has been resolved by add a LatestOnlyOperator in #1752