**Apache Airflow version**: 2.1.0

**What happened**:

We use a python script that reads JSON configs from db to dynamically generate DAGs. Sometimes, when JSON Configs is updated, we expect DAGs to be removed in Airflow.  This doesn't always happen. From my observation, one of the following happens:
1) DAG is removed because the python script no longer generates it.
2) DAG still exists but run history is empty. When trigger the DAG, first task is stuck in `queued` status indefinitely.

**What you expected to happen**:
I expect the right behavior should be 
>1) DAG is removed because the python script no longer generates it.

**How to reproduce it**:
* Create a python script in DAG folder that dynamically generates multiple DAGs
* Execute these dynamically generated DAGs a few times
* Use Airflow Variable to toggle (reduce) the # of DAGs generated
* Examine if number of dynamically generated DAGs in the web UI


airflow/models/dag.py
)
        most_recent_runs_iter = session.query(DagRun).filter(
            DagRun.dag_id == most_recent_subq.c.dag_id,
            DagRun.execution_date == most_recent_subq.c.max_execution_date,
        )
        most_recent_runs = {run.dag_id: run for run in most_recent_runs_iter}
        filelocs = []
        for orm_dag in sorted(orm_dags, key=lambda d: d.dag_id):
            dag = dag_by_ids[orm_dag.dag_id]
            filelocs.append(dag.fileloc)
            if dag.is_subdag:
                orm_dag.is_subdag = True
                orm_dag.fileloc = dag.parent_dag.fileloc  
                orm_dag.root_dag_id = dag.parent_dag.dag_id  
                orm_dag.owners = dag.parent_dag.owner  
            else:
                orm_dag.is_subdag = False
                orm_dag.fileloc = dag.fileloc
                orm_dag.owners = dag.owner
            orm_dag.is_active = True
            orm_dag.last_parsed_time = timezone.utcnow()
            orm_dag.default_view = dag.default_view
            orm_dag.description = dag.description
            orm_dag.schedule_interval = dag.schedule_interval
            orm_dag.max_active_tasks = dag.max_active_tasks
            orm_dag.max_active_runs = dag.max_active_runs
            orm_dag.has_task_concurrency_limits = any(t.max_active_tis_per_dag is not None for t in dag.tasks)
            run: Optional[DagRun] = most_recent_runs.get(dag.dag_id)
            if run is None:
                data_interval = None