diff --git a/homeassistant/components/logbook/__init__.py b/homeassistant/components/logbook/__init__.py
index de0f901be3b..87b66c57730 100644
--- a/homeassistant/components/logbook/__init__.py
+++ b/homeassistant/components/logbook/__init__.py
@@ -500,10 +500,10 @@ def _generate_events_query(session):
 def _generate_events_query_without_states(session):
     return session.query(
         *EVENT_COLUMNS,
-        literal(None).label("state"),
-        literal(None).label("entity_id"),
-        literal(None).label("domain"),
-        literal(None).label("attributes"),
+        literal(value=None, type_=sqlalchemy.String).label("state"),
+        literal(value=None, type_=sqlalchemy.String).label("entity_id"),
+        literal(value=None, type_=sqlalchemy.String).label("domain"),
+        literal(value=None, type_=sqlalchemy.Text).label("attributes"),
     )
 
 
diff --git a/homeassistant/components/recorder/migration.py b/homeassistant/components/recorder/migration.py
index 51d0f0a86fd..8e6c4861739 100644
--- a/homeassistant/components/recorder/migration.py
+++ b/homeassistant/components/recorder/migration.py
@@ -312,6 +312,34 @@ def _update_states_table_with_foreign_key_options(connection, engine):
             )
 
 
+def _drop_foreign_key_constraints(connection, engine, table, columns):
+    """Drop foreign key constraints for a table on specific columns."""
+    inspector = sqlalchemy.inspect(engine)
+    drops = []
+    for foreign_key in inspector.get_foreign_keys(table):
+        if (
+            foreign_key["name"]
+            and foreign_key["options"].get("ondelete")
+            and foreign_key["constrained_columns"] == columns
+        ):
+            drops.append(ForeignKeyConstraint((), (), name=foreign_key["name"]))
+
+    # Bind the ForeignKeyConstraints to the table
+    old_table = Table(  # noqa: F841 pylint: disable=unused-variable
+        table, MetaData(), *drops
+    )
+
+    for drop in drops:
+        try:
+            connection.execute(DropConstraint(drop))
+        except (InternalError, OperationalError):
+            _LOGGER.exception(
+                "Could not drop foreign constraints in %s table on %s",
+                TABLE_STATES,
+                columns,
+            )
+
+
 def _apply_update(engine, session, new_version, old_version):
     """Perform operations to bring schema up to date."""
     connection = session.connection()
@@ -420,6 +448,10 @@ def _apply_update(engine, session, new_version, old_version):
             # Recreate the statistics table
             Statistics.__table__.drop(engine)
             Statistics.__table__.create(engine)
+    elif new_version == 16:
+        _drop_foreign_key_constraints(
+            connection, engine, TABLE_STATES, ["old_state_id"]
+        )
     else:
         raise ValueError(f"No schema migration defined for version {new_version}")
 
diff --git a/homeassistant/components/recorder/models.py b/homeassistant/components/recorder/models.py
index 924f59790b0..4fefcaa19e3 100644
--- a/homeassistant/components/recorder/models.py
+++ b/homeassistant/components/recorder/models.py
@@ -8,6 +8,7 @@ from sqlalchemy import (
     DateTime,
     Float,
     ForeignKey,
+    Identity,
     Index,
     Integer,
     String,
@@ -28,7 +29,7 @@ import homeassistant.util.dt as dt_util
 # pylint: disable=invalid-name
 Base = declarative_base()
 
-SCHEMA_VERSION = 15
+SCHEMA_VERSION = 16
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -61,7 +62,7 @@ class Events(Base):  # type: ignore
         "mysql_collate": "utf8mb4_unicode_ci",
     }
     __tablename__ = TABLE_EVENTS
-    event_id = Column(Integer, primary_key=True)
+    event_id = Column(Integer, Identity(), primary_key=True)
     event_type = Column(String(MAX_LENGTH_EVENT_TYPE))
     event_data = Column(Text().with_variant(mysql.LONGTEXT, "mysql"))
     origin = Column(String(32))
@@ -128,7 +129,7 @@ class States(Base):  # type: ignore
         "mysql_collate": "utf8mb4_unicode_ci",
     }
     __tablename__ = TABLE_STATES
-    state_id = Column(Integer, primary_key=True)
+    state_id = Column(Integer, Identity(), primary_key=True)
     domain = Column(String(64))
     entity_id = Column(String(255))
     state = Column(String(255))
@@ -139,9 +140,7 @@ class States(Base):  # type: ignore
     last_changed = Column(DATETIME_TYPE, default=dt_util.utcnow)
     last_updated = Column(DATETIME_TYPE, default=dt_util.utcnow, index=True)
     created = Column(DATETIME_TYPE, default=dt_util.utcnow)
-    old_state_id = Column(
-        Integer, ForeignKey("states.state_id", ondelete="NO ACTION"), index=True
-    )
+    old_state_id = Column(Integer, ForeignKey("states.state_id"), index=True)
     event = relationship("Events", uselist=False)
     old_state = relationship("States", remote_side=[state_id])
 
@@ -246,7 +245,7 @@ class RecorderRuns(Base):  # type: ignore
     """Representation of recorder run."""
 
     __tablename__ = TABLE_RECORDER_RUNS
-    run_id = Column(Integer, primary_key=True)
+    run_id = Column(Integer, Identity(), primary_key=True)
     start = Column(DateTime(timezone=True), default=dt_util.utcnow)
     end = Column(DateTime(timezone=True))
     closed_incorrect = Column(Boolean, default=False)
@@ -297,7 +296,7 @@ class SchemaChanges(Base):  # type: ignore
     """Representation of schema version changes."""
 
     __tablename__ = TABLE_SCHEMA_CHANGES
-    change_id = Column(Integer, primary_key=True)
+    change_id = Column(Integer, Identity(), primary_key=True)
     schema_version = Column(Integer)
     changed = Column(DateTime(timezone=True), default=dt_util.utcnow)