Option to include subdir in trigger_dag to not make scheduler scan the whole dag folder

[GagandeepS]

We have a design where 10s of child dags gets created every few minutes and gets triggered from a parent dag. While triggering each of the child dag, I believe scheduler searches the whole dag bag which is making the whole process slower. We have put a while loop to run TriggerDagRunOperator/trigger_dagy, if it is successful in triggering then it exits the loop otherwise trigger it again.

I believe that to decrease the load from the scheduler, there should be a provision to supply subdir to TriggerDagRunOperator so that scheduler only searched for the dag_id inside of that folder instead of whole dagbag.

airflow/airflow/api/common/experimental/trigger_dag.py

Line 117 in 37a12e9

dagbag = DagBag(dag_folder=dag_model.fileloc, read_dags_from_db=True)