**Apache Airflow version**: 2.0.1


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

**Environment**:

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

**What happened**:

Airflow fails to schedule any tasks after some time. The "Task Instance Details" tab of some of the failed tasks show the following: 

```
('Not scheduling since there are %s open slots in pool %s and require %s pool slots', -3, 'transformation', 3)
```

Admin > Pools tab shows 0 Running Slots but 9 Queued Slots. Gets stuck in this state until airflow is restarted.

**What you expected to happen**:

Number of "open slots in pool" should never be negative! 

**How to reproduce it**:
- Create/configure a pool with a small size (eg. 6)
- DAG with multiple tasks occupying multiple pool_slots (eg. pool_slots=3)


**Anything else we need to know**:

airflow/models/pool.py
*,
        lock_rows: bool = False,
        session: Session = None,
    ) -> Dict[str, PoolStats]:
        """
        Get Pool stats (Number of Running, Queued, Open & Total tasks)
        If ``lock_rows`` is True, and the database engine in use supports the ``NOWAIT`` syntax, then a
        non-blocking lock will be attempted -- if the lock is not available then SQLAlchemy will throw an
        OperationalError.
        :param lock_rows: Should we attempt to obtain a row-level lock on all the Pool rows returns
        :param session: SQLAlchemy ORM Session
        """
        from airflow.models.taskinstance import TaskInstance  
        pools: Dict[str, PoolStats] = {}
        query = session.query(Pool.pool, Pool.slots)
        if lock_rows:
            query = with_row_locks(query, session=session, **nowait(session))
        pool_rows: Iterable[Tuple[str, int]] = query.all()
        for (pool_name, total_slots) in pool_rows:
            pools[pool_name] = PoolStats(total=total_slots, running=0, queued=0, open=0)
        state_count_by_pool = (
            session.query(TaskInstance.pool, TaskInstance.state, func.count())
            .filter(TaskInstance.state.in_(list(EXECUTION_STATES)))
            .group_by(TaskInstance.pool, TaskInstance.state)
        ).all()
        
        count: int
        for (pool_name, state, count) in state_count_by_pool:
            stats_dict: Optional[PoolStats] = pools.get(pool_name)