https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
self, task_instances: list[TaskInstance], session: Session, clear_downstream: bool = False
    ) -> tuple[int, int]:
        """
        Clear task instances, optionally including their downstream dependencies.
        :param task_instances: list of TIs to clear
        :param clear_downstream: should downstream task instances be cleared as well?
        :return: a tuple with:
            - count of cleared task instances actually selected by the user
            - count of downstream task instances that were additionally cleared
        """
        cleared_tis_count = 0
        cleared_downstream_tis_count = 0
        
        tis_grouped_by_dag_id = itertools.groupby(task_instances, lambda ti: ti.dag_id)
        for dag_id, dag_tis in tis_grouped_by_dag_id:
            dag = get_airflow_app().dag_bag.get_dag(dag_id)
            tis_to_clear = list(dag_tis)
            downstream_tis_to_clear = []
            if clear_downstream:
                tis_to_clear_grouped_by_dag_run = itertools.groupby(tis_to_clear, lambda ti: ti.dag_run)
                for dag_run, dag_run_tis in tis_to_clear_grouped_by_dag_run:
                    
                    
                    task_ids_to_clear = [ti.task_id for ti in dag_run_tis]
                    partial_dag = dag.partial_subset(
                        task_ids_or_regex=task_ids_to_clear, include_downstream=True, include_upstream=False
                    )
                    downstream_task_ids_to_clear = [
                        task_id for task_id in partial_dag.task_dict if task_id not in task_ids_to_clear
                    ]