https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
info = dag_run.task_instance_scheduling_decisions(session)
            skippable_task_ids = {
                task_id for task_id in partial_dag.task_ids if task_id not in task.downstream_task_ids
            }
            schedulable_tis = [
                ti
                for ti in info.schedulable_tis
                if ti.task_id not in skippable_task_ids
                and not (
                    ti.task.inherits_from_empty_operator
                    and not ti.task.on_execute_callback
                    and not ti.task.on_success_callback
                    and not ti.task.outlets
                )
            ]
            for schedulable_ti in schedulable_tis:
                if not hasattr(schedulable_ti, "task"):
                    schedulable_ti.task = task.dag.get_task(schedulable_ti.task_id)
            num = dag_run.schedule_tis(schedulable_tis, session=session, max_tis_per_query=max_tis_per_query)
            self.log.info("%d downstream tasks scheduled from follow-on schedule check", num)
            session.flush()
        except OperationalError as e:
            
            self.log.info(
                "Skipping mini scheduling run due to exception: %s",
                e.statement,
                exc_info=True,
            )
            session.rollback()
    def get_relevant_upstream_map_indexes(