Clear tasks by task ids in REST API

[vemikhaylov]

closes: #13225

[turbaszek]

How this will work with

 tis = tis.filter(TI.task_id.in_(self.task_ids))

from L1197?

[vemikhaylov]

Actually the conditions are just added with conjunction:

# tst_double_in_query.py
from sqlalchemy.orm import Session

from airflow.models import TaskInstance

session = Session()
query = session.query(TaskInstance.task_id).filter(TaskInstance.task_id.in_(["foo"]))
print(str(query.statement))
query = query.filter(TaskInstance.task_id.in_(["bar"]))
print(str(query.statement))

$ python tst_double_in_query.py
First query:
SELECT task_instance.task_id
FROM task_instance
WHERE task_instance.task_id IN (:task_id_1)

Second query:
SELECT task_instance.task_id
FROM task_instance
WHERE task_instance.task_id IN (:task_id_1) AND task_instance.task_id IN (:task_id_2)

So the second filter narrows down the search space if task_ids are provided.

Naturally we can intersect the sets preliminary and apply the filter once, it can make the generated SQL code a little more efficient. Would it be better, how do you feel?

[vemikhaylov]

I'm not sure whether we should check provided task_ids and raise the 400 or the 404 error if any of them doesn't exist for the given DAG or just apply the filter blindly.

[ephraimbuddy]

LGTM @mik-laj @turbaszek

[github-actions]

The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest master at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.

[ephraimbuddy]

Please rebase and push

[kurtqq]

@kaxil @ephraimbuddy gentle ping

[ephraimbuddy]

Closing and reopening to trigger test runs