[AIRFLOW-5391] Do not run skipped tasks when they are cleared

[yuqian90]

This PR fixes the following issue:

If a task is skipped by BranchPythonOperator, BaseBranchOperator orr ShortCircuitOperator and the user then clears the skipped task, it'll execute.

After this PR:
The NotPreviouslySkippedDep rule will first evaluate if a task has a direct SkipMixin parent that has decided to skip it. This is done by examining the XCom data stored by SkipMixin.skip() or SkipMixin.skip_all_except().

The implementation is inspired by the author of this blog.

---

Issue link: AIRFLOW-5391

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

• Description above provides context of the change
• Commit message/PR title starts with [AIRFLOW-NNNN]. AIRFLOW-NNNN = JIRA ID^*
• Unit tests coverage for changes (not needed for documentation changes)
• 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.

^* For document-only changes commit message can start with [AIRFLOW-XXXX].

---

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.

[tooptoop4]

hmm, I use BranchPythonOperator and clear command a lot and I never see the skipped tasks just run on clear. do u use none_failed ?

[yuqian90]

@tooptoop4 the problem happens when you clear the skipped task (not when you clear the BranchPythonOperator). There's an example in the linked JIRA.

[tooptoop4]

i clear entire dag and dont face problem

[yuqian90]

@tooptoop4 yes if always clear the entire DAG, the BranchPythonOperator itself gets cleared so you won't face this problem.

[tooptoop4]

makes sense!

[codecov-io]

Codecov Report

Merging #7276 into master will increase coverage by 53.43%.
The diff coverage is 100%.

@@             Coverage Diff             @@
##           master    #7276       +/-   ##
===========================================
+ Coverage   32.95%   86.39%   +53.43%     
===========================================
  Files         878      878               
  Lines       41219    41288       +69     
===========================================
+ Hits        13584    35669    +22085     
+ Misses      27635     5619    -22016

Impacted Files	Coverage Δ
airflow/ti_deps/deps/trigger_rule_dep.py	91.02% <ø> (+74.77%)	⬆️
airflow/models/baseoperator.py	96.52% <100%> (+38.94%)	⬆️
airflow/ti_deps/dep_context.py	100% <100%> (+29.03%)	⬆️
airflow/models/skipmixin.py	98.18% <100%> (+73.79%)	⬆️
airflow/ti_deps/deps/not_previously_skipped_dep.py	100% <100%> (ø)
airflow/kubernetes/volume_mount.py	44.44% <0%> (-55.56%)	⬇️
airflow/kubernetes/volume.py	52.94% <0%> (-47.06%)	⬇️
airflow/kubernetes/pod_launcher.py	47.18% <0%> (-39.44%)	⬇️
...viders/cncf/kubernetes/operators/kubernetes_pod.py	69.38% <0%> (-25.52%)	⬇️
airflow/kubernetes/refresh_config.py	50.98% <0%> (-23.53%)	⬇️
... and 767 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update d83ce26...39020d8. Read the comment docs.

[yuqian90]

• Added tests
• Extended the same support to BranchPythonOperator, ShortCircuitOperator and BaseBranchOperator. Clearing downstream tasks of all these three operators will now do the right thing.
• That said, I still think BaseBranchOperator is a bit redundant since we already have BranchPythonOperator. I'm proposing to deprecate BaesBranchOperator in another PR

The PR is now ready for review. Reviewers @kaxil @ashb @feluelle please take a look.

[yuqian90]

@ashb @kaxil @feluelle @tooptoop4 Hoping to get a review here.

I have updated the PR a bit. The logic is still the same, but the code that does the branching evaluation has been moved into a new dependency rule inside ti_deps/deps/branch_dep.py.

When BranchDep is evaluated, it sets the state of a task to "SKIPPED" if it has a parent that decided to skip it.

For example, this comes from "example_short_circuit_operatorschedule":

For a DAG that looks like this, condition_is_False decided to skip false_1. Before this PR, the problem is if someone clears false_1, it'll execute. This is very counter-intuitive for users:

After this PR, when the scheduler evaluates the dependency rules for false_1, BranchDep will skip false_1 because it knows that the parent task condition_is_False had already decided to skip false_1. In other words, this makes the "skipped" status "sticky".

So after this PR, the DAG will look like this after someone clears false_1. This is much more intuitive:

[feluelle]

Overall LGTM. Great test coverage 👍

..would be nice if you could complete the docstrings to contain types and add the types also to the function header.

[yuqian90]

Hi, @feluelle I've adopted all your suggestions (adding doc and raise ValueError), and the PR has been updated. Please take another look. Thank you!

[ashb]

Okay, why you are using XCom makes more sense now but I still think it should use a non-default Key (as skip Mixin could be used elsewhere, and we don't want to overwrite possible xcom from the task)