**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/jobs/scheduler_job.py
self.log.exception("Could not send SLA Miss email notification for DAG %s", dag.dag_id)
            
            if notification_sent:
                for sla in slas:
                    sla.email_sent = email_sent
                    sla.notification_sent = True
                    session.merge(sla)
            session.commit()
    @staticmethod
    def update_import_errors(session: Session, dagbag: DagBag) -> None:
        """
        For the DAGs in the given DagBag, record any associated import errors and clears
        errors for files that no longer have them. These are usually displayed through the
        Airflow UI so that users know that there are issues parsing DAGs.
        :param session: session for ORM operations
        :type session: sqlalchemy.orm.session.Session
        :param dagbag: DagBag containing DAGs with import errors
        :type dagbag: airflow.DagBag
        """
        
        for dagbag_file in dagbag.file_last_changed:
            session.query(errors.ImportError).filter(errors.ImportError.filename == dagbag_file).delete()
        
        for filename, stacktrace in dagbag.import_errors.items():
            session.add(
                errors.ImportError(filename=filename, timestamp=timezone.utcnow(), stacktrace=stacktrace)
            )
        session.commit()
    @provide_session
    def execute_callbacks(