**Apache Airflow version**: 2.0.0rc3

**Kubernetes version (if you are using kubernetes)** (use `kubectl version`):

**Environment**:

- **Cloud provider or hardware configuration**: Linux
- **OS** (e.g. from /etc/os-release): Ubuntu
- **Kernel** (e.g. `uname -a`): 
- **Install tools**:
- **Others**:

**What happened**:

After shutting down the scheduler, while tasks were in running state, trying to restart the scheduler results in pk violations..

```[2020-12-15 22:43:29,673] {scheduler_job.py:1293} ERROR - Exception when executing SchedulerJob._run_scheduler_loop
Traceback (most recent call last):
  File "/home/jcoder/git/airflow_2.0/pyenv/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1277, in _execute_context
    cursor, statement, parameters, context
  File "/home/jcoder/git/airflow_2.0/pyenv/lib/python3.7/site-packages/sqlalchemy/engine/default.py", line 593, in do_execute
    cursor.execute(statement, parameters)
psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint "dag_run_dag_id_run_id_key"
DETAIL:  Key (dag_id, run_id)=(example_task_group, scheduled__2020-12-14T04:31:00+00:00) already exists.
```


**What you expected to happen**:

Scheduler restarts and picks up where it left off.

**How to reproduce it**:

Set example dag ( I used task_group) to schedule_interval `* * * * *` and start the scheduler and let it run for a few minutes. 
Shut down the scheduler
Attempt to restart the scheduler

**Anything else we need to know**:
I came across this doing testing using the LocalExecutor in a virtual env. If no else is able to reproduce it, I'll try again in a clean virtual env.

airflow/models/dagbag.py
"""Information about single file"""
    file: str
    duration: timedelta
    dag_num: int
    task_num: int
    dags: str
class DagBag(LoggingMixin):
    """
    A dagbag is a collection of dags, parsed out of a folder tree and has high
    level configuration settings, like what database to use as a backend and
    what executor to use to fire off tasks. This makes it easier to run
    distinct environments for say production and development, tests, or for
    different teams or security profiles. What would have been system level
    settings are now dagbag level so that one system can run multiple,
    independent settings sets.
    :param dag_folder: the folder to scan to find DAGs
    :type dag_folder: unicode
    :param include_examples: whether to include the examples that ship
        with airflow or not
    :type include_examples: bool
    :param include_smart_sensor: whether to include the smart sensor native
        DAGs that create the smart sensor operators for whole cluster
    :type include_smart_sensor: bool
    :param read_dags_from_db: Read DAGs from DB if ``True`` is passed.
        If ``False`` DAGs are read from python files.
    :type read_dags_from_db: bool
    """
    DAGBAG_IMPORT_TIMEOUT = conf.getfloat('core', 'DAGBAG_IMPORT_TIMEOUT')
    SCHEDULER_ZOMBIE_TASK_THRESHOLD = conf.getint('scheduler', 'scheduler_zombie_task_threshold')
    def __init__(