### Apache Airflow version

2.2.0b2 (beta snapshot)

### Operating System

Debian GNU/Linux 11 (bullseye)

### Versions of Apache Airflow Providers

```
apache-airflow-providers-amazon @ file:///root/.cache/pypoetry/artifacts/c9/69/16/ffa2eb7a2e6e850a7048eaf66b6c40c990ef7c58149f20d3d3f333a2e9/apache_airflow_providers_amazon-2.2.0-py3-none-any.whl
apache-airflow-providers-celery @ file:///root/.cache/pypoetry/artifacts/6e/1b/2f/f968318a7474e979af4dc53893ecafe8cd11a98a94077a9c3c27304eb7/apache_airflow_providers_celery-2.1.0-py3-none-any.whl
apache-airflow-providers-ftp @ file:///root/.cache/pypoetry/artifacts/8b/9a/dd/79a36c62bc7f37f98d0ea33652570e19272e8a7a2297db13a6785698d1/apache_airflow_providers_ftp-2.0.1-py3-none-any.whl
apache-airflow-providers-http @ file:///root/.cache/pypoetry/artifacts/52/28/81/03a89147daf7daceb55f1218189d1c4af01c33c45849b568769ca6765f/apache_airflow_providers_http-2.0.1-py3-none-any.whl
apache-airflow-providers-imap @ file:///root/.cache/pypoetry/artifacts/1c/5d/c5/269e8a8098e7017a26a2a376eb3020e1a864775b7ff310ed39e1bd503d/apache_airflow_providers_imap-2.0.1-py3-none-any.whl
apache-airflow-providers-postgres @ file:///root/.cache/pypoetry/artifacts/fb/69/ac/e8e25a0f6a4b0daf162c81c9cfdbb164a93bef6bd652c1c00eee6e0815/apache_airflow_providers_postgres-2.3.0-py3-none-any.whl
apache-airflow-providers-redis @ file:///root/.cache/pypoetry/artifacts/cf/2b/56/75563b6058fe45b70f93886dd92541e8349918eeea9d70c703816f2639/apache_airflow_providers_redis-2.0.1-py3-none-any.whl
apache-airflow-providers-sqlite @ file:///root/.cache/pypoetry/artifacts/61/ba/e9/c0b4b7ef2599dbd902b32afc99f2620d8a616b3072122e90f591de4807/apache_airflow_providers_sqlite-2.0.1-py3-none-any.whl
```

### Deployment

Other Docker-based deployment

### Deployment details

AWS ECS + Fargate, RabbitMQ, Celery executor, Postgres 13 backend

### What happened

Hey so we just hit an issue with the 2.2.0b2 migration `airflow/migrations/versions/7b2661a43ba3_taskinstance_keyed_to_dagrun.py`
 . We have dag_runs without a run_id (as it is still nullable) on the dag_run table. This caused the FK constraint on TI not to apply.
Noticeably this wasn't caught by https://github.com/apache/airflow/blob/ca93ff4ef05e68369b4fd42fcda3b40f63682d4a/airflow/utils/db.py#L700 before the migrations apply. Is it worth adding an additional check here and making dag_run.run_id NOT NULL?

### What you expected to happen

The data check before applying migrations should have spotted the NULL run_ids on the dag_run table. Going forward perhaps we can make this column NOT NULL?

### How to reproduce

NULL some run_id columns on the dag_run table. As to how this occurs I am not sure.

### Anything else

_No response_

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

tests/api_connexion/schemas/test_dag_run_schema.py
"""Dag_run_id and execution_date fields are autogenerated if missing"""
        serialized_dagrun = {}
        result = dagrun_schema.load(serialized_dagrun)
        assert result == {"execution_date": result["execution_date"], "run_id": result["run_id"]}
    def test_invalid_execution_date_raises(self):
        serialized_dagrun = {"execution_date": "mydate"}
        with pytest.raises(BadRequest) as ctx:
            dagrun_schema.load(serialized_dagrun)
        assert str(ctx.value) == "Incorrect datetime argument"
class TestDagRunCollection(TestDAGRunBase):
    @provide_session
    def test_serialize(self, session):
        dagrun_model_1 = DagRun(
            run_id="my-dag-run",
            state='running',
            execution_date=timezone.parse(self.default_time),
            run_type=DagRunType.MANUAL.value,
            start_date=timezone.parse(self.default_time),
            conf='{"start": "stop"}',
        )
        dagrun_model_2 = DagRun(
            run_id="my-dag-run-2",
            state='running',
            execution_date=timezone.parse(self.second_time),
            start_date=timezone.parse(self.default_time),
            run_type=DagRunType.MANUAL.value,
        )
        dagruns = [dagrun_model_1, dagrun_model_2]
        session.add_all(dagruns)
        session.commit()