Separate attrs into another table (reduces database size)

homeassistant/components/logbook/__init__.py

@@ -5,6 +5,7 @@
 from itertools import groupby
 import json
 import re
+from typing import Any
 
 import sqlalchemy
 from sqlalchemy.orm import aliased
@@ -17,6 +18,7 @@
 from homeassistant.components.http import HomeAssistantView
 from homeassistant.components.recorder.models import (
     Events,
+    StateAttributes,
     States,
     process_timestamp_to_utc_isoformat,
 )
@@ -493,6 +495,7 @@ def _generate_events_query(session):
         States.entity_id,
         States.domain,
         States.attributes,
+        StateAttributes.shared_attrs,
     )
 
 
@@ -503,6 +506,7 @@ def _generate_events_query_without_states(session):
         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"),
+        literal(value=None, type_=sqlalchemy.Text).label("shared_attrs"),
     )
 
 
@@ -518,6 +522,9 @@ def _generate_states_query(session, start_day, end_day, old_state, entity_ids):
             (States.last_updated == States.last_changed)
             & States.entity_id.in_(entity_ids)
         )
+        .outerjoin(
+            StateAttributes, (States.attributes_id == StateAttributes.attributes_id)
+        )
     )
 
 
@@ -533,7 +540,9 @@ def _apply_events_types_and_states_filter(hass, query, old_state):
             (Events.event_type != EVENT_STATE_CHANGED) | _continuous_entity_matcher()
         )
     )
-    return _apply_event_types_filter(hass, events_query, ALL_EVENT_TYPES)
+    return _apply_event_types_filter(hass, events_query, ALL_EVENT_TYPES).outerjoin(
+        StateAttributes, (States.attributes_id == StateAttributes.attributes_id)
+    )
 
 
 def _missing_state_matcher(old_state):
@@ -555,6 +564,9 @@ def _continuous_entity_matcher():
     return sqlalchemy.or_(
         sqlalchemy.not_(States.domain.in_(CONTINUOUS_DOMAINS)),
         sqlalchemy.not_(States.attributes.contains(UNIT_OF_MEASUREMENT_JSON)),
+        sqlalchemy.not_(
+            StateAttributes.shared_attrs.contains(UNIT_OF_MEASUREMENT_JSON)
+        ),
     )
 
 
@@ -708,8 +720,9 @@ def attributes_icon(self):
         """Extract the icon from the decoded attributes or json."""
         if self._attributes:
             return self._attributes.get(ATTR_ICON)
-
-        result = ICON_JSON_EXTRACT.search(self._row.attributes)
+        result = ICON_JSON_EXTRACT.search(
+            self._row.shared_attrs or self._row.attributes
+        )
         return result and result.group(1)
 
     @property
@@ -733,14 +746,12 @@ def data_domain(self):
     @property
     def attributes(self):
         """State attributes."""
-        if not self._attributes:
-            if (
-                self._row.attributes is None
-                or self._row.attributes == EMPTY_JSON_OBJECT
-            ):
+        if self._attributes is None:
+            source = self._row.shared_attrs or self._row.attributes
+            if source == EMPTY_JSON_OBJECT or source is None:
                 self._attributes = {}
             else:
-                self._attributes = json.loads(self._row.attributes)
+                self._attributes = json.loads(source) or {}
         return self._attributes
 
     @property
@@ -771,12 +782,12 @@ class EntityAttributeCache:
     that are expected to change state.
     """
 
-    def __init__(self, hass):
+    def __init__(self, hass: HomeAssistant) -> None:
         """Init the cache."""
         self._hass = hass
-        self._cache = {}
+        self._cache: dict[str, dict[str, Any]] = {}
 
-    def get(self, entity_id, attribute, event):
+    def get(self, entity_id: str, attribute: str, event: LazyEventPartialState) -> Any:
         """Lookup an attribute for an entity or get it from the cache."""
         if entity_id in self._cache:
             if attribute in self._cache[entity_id]:

homeassistant/components/recorder/__init__.py

@@ -14,6 +14,7 @@
 import time
 from typing import Any
 
+from lru import LRU
 from sqlalchemy import create_engine, event as sqlalchemy_event, exc, func, select
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.orm import scoped_session, sessionmaker
@@ -65,6 +66,7 @@
     Base,
     Events,
     RecorderRuns,
+    StateAttributes,
     States,
     StatisticsRuns,
     process_timestamp,
@@ -520,6 +522,8 @@ def __init__(
         self._commits_without_expire = 0
         self._keepalive_count = 0
         self._old_states: dict[str, States] = {}
+        self._state_attributes_ids: LRU = LRU(2048)
+        self._pending_state_attributes: dict[str, StateAttributes] = {}
         self._pending_expunge: list[States] = []
         self.event_session = None
         self.get_session = None
@@ -910,34 +914,70 @@ def _process_one_event(self, event):
                 dbevent.event_data = None
             else:
                 dbevent = Events.from_event(event)
-            self.event_session.add(dbevent)
         except (TypeError, ValueError):
             _LOGGER.warning("Event is not JSON serializable: %s", event)
             return
 
+        self.event_session.add(dbevent)
         if event.event_type == EVENT_STATE_CHANGED:
             try:
                 dbstate = States.from_event(event)
-                has_new_state = event.data.get("new_state")
-                if dbstate.entity_id in self._old_states:
-                    old_state = self._old_states.pop(dbstate.entity_id)
-                    if old_state.state_id:
-                        dbstate.old_state_id = old_state.state_id
-                    else:
-                        dbstate.old_state = old_state
-                if not has_new_state:
-                    dbstate.state = None
-                dbstate.event = dbevent
-                self.event_session.add(dbstate)
-                if has_new_state:
-                    self._old_states[dbstate.entity_id] = dbstate
-                    self._pending_expunge.append(dbstate)
-            except (TypeError, ValueError):
+                dbstate.attributes = None
+                dbstate_attributes = StateAttributes.from_event(event)
+            except (TypeError, ValueError) as ex:
                 _LOGGER.warning(
-                    "State is not JSON serializable: %s",
+                    "State is not JSON serializable: %s: %s",
                     event.data.get("new_state"),
+                    ex,
                 )
 
+            save_attributes = True
+            attributes_id = None
+            if pending_attributes := self._pending_state_attributes.get(
+                dbstate_attributes.shared_attrs
+            ):
+                dbstate_attributes = pending_attributes
+                save_attributes = False
+            elif attributes_id := self._state_attributes_ids.get(
+                dbstate_attributes.shared_attrs
+            ):
+                save_attributes = False
+            elif (
+                attributes := self.event_session.query(StateAttributes.attributes_id)
+                .filter(StateAttributes.hash == dbstate_attributes.hash)
+                .filter(StateAttributes.shared_attrs == dbstate_attributes.shared_attrs)
+                .first()
+            ):
+                attributes_id = attributes[0]
+                self._state_attributes_ids[
+                    dbstate_attributes.shared_attrs
+                ] = attributes_id
+                save_attributes = False
+
+            has_new_state = event.data.get("new_state")
+            if dbstate.entity_id in self._old_states:
+                old_state = self._old_states.pop(dbstate.entity_id)
+                if old_state.state_id:
+                    dbstate.old_state_id = old_state.state_id
+                else:
+                    dbstate.old_state = old_state
+            if not has_new_state:
+                dbstate.state = None
+            dbstate.event = dbevent
+            self.event_session.add(dbstate)
+            if attributes_id:
+                dbstate.attributes_id = attributes_id
+            else:
+                dbstate.state_attributes = dbstate_attributes
+            if save_attributes:
+                self._pending_state_attributes[
+                    dbstate_attributes.shared_attrs
+                ] = dbstate_attributes
+                self.event_session.add(dbstate_attributes)
+            if has_new_state:
+                self._old_states[dbstate.entity_id] = dbstate
+                self._pending_expunge.append(dbstate)
+
         # If they do not have a commit interval
         # than we commit right away
         if not self.commit_interval:
@@ -986,6 +1026,7 @@ def _commit_event_session(self):
                 if dbstate in self.event_session:
                     self.event_session.expunge(dbstate)
             self._pending_expunge = []
+        self._pending_state_attributes = {}
         self.event_session.commit()
 
         # Expire is an expensive operation (frequently more expensive
@@ -1006,6 +1047,8 @@ def _handle_sqlite_corruption(self):
     def _close_event_session(self):
         """Close the event session."""
         self._old_states = {}
+        self._state_attributes_ids = {}
+        self._pending_state_attributes = {}
 
         if not self.event_session:
             return

homeassistant/components/recorder/history.py

@@ -6,14 +6,20 @@
 import logging
 import time
 
+import sqlalchemy
 from sqlalchemy import and_, bindparam, func
 from sqlalchemy.ext import baked
 
 from homeassistant.components import recorder
 from homeassistant.core import split_entity_id
 import homeassistant.util.dt as dt_util
 
-from .models import LazyState, States, process_timestamp_to_utc_isoformat
+from .models import (
+    LazyState,
+    StateAttributes,
+    States,
+    process_timestamp_to_utc_isoformat,
+)
 from .util import execute, session_scope
 
 # mypy: allow-untyped-defs, no-check-untyped-defs
@@ -46,6 +52,7 @@
     States.attributes,
     States.last_changed,
     States.last_updated,
+    StateAttributes.shared_attrs,
 ]
 
 HISTORY_BAKERY = "recorder_history_bakery"
@@ -86,11 +93,9 @@ def get_significant_states_with_session(
     thermostat so that we get current temperature in our graphs).
     """
     timer_start = time.perf_counter()
-
     baked_query = hass.data[HISTORY_BAKERY](
         lambda session: session.query(*QUERY_STATES)
     )
-
     if significant_changes_only:
         baked_query += lambda q: q.filter(
             (
@@ -113,7 +118,9 @@ def get_significant_states_with_session(
 
     if end_time is not None:
         baked_query += lambda q: q.filter(States.last_updated < bindparam("end_time"))
-
+    baked_query += lambda q: q.outerjoin(
+        StateAttributes, States.attributes_id == StateAttributes.attributes_id
+    )
     baked_query += lambda q: q.order_by(States.entity_id, States.last_updated)
 
     states = execute(
@@ -159,6 +166,9 @@ def state_changes_during_period(hass, start_time, end_time=None, entity_id=None)
             baked_query += lambda q: q.filter_by(entity_id=bindparam("entity_id"))
             entity_id = entity_id.lower()
 
+        baked_query += lambda q: q.outerjoin(
+            StateAttributes, States.attributes_id == StateAttributes.attributes_id
+        )
         baked_query += lambda q: q.order_by(States.entity_id, States.last_updated)
 
         states = execute(
@@ -185,7 +195,9 @@ def get_last_state_changes(hass, number_of_states, entity_id):
         if entity_id is not None:
             baked_query += lambda q: q.filter_by(entity_id=bindparam("entity_id"))
             entity_id = entity_id.lower()
-
+        baked_query += lambda q: q.outerjoin(
+            StateAttributes, States.attributes_id == StateAttributes.attributes_id
+        )
         baked_query += lambda q: q.order_by(
             States.entity_id, States.last_updated.desc()
         )
@@ -194,7 +206,8 @@ def get_last_state_changes(hass, number_of_states, entity_id):
 
         states = execute(
             baked_query(session).params(
-                number_of_states=number_of_states, entity_id=entity_id
+                number_of_states=number_of_states,
+                entity_id=entity_id,
             )
         )
 
@@ -243,7 +256,7 @@ def _get_states_with_session(
 
     # We have more than one entity to look at so we need to do a query on states
     # since the last recorder run started.
-    query = session.query(*QUERY_STATES)
+    query: sqlalchemy.orm.Query = session.query(*QUERY_STATES)
 
     if entity_ids:
         # We got an include-list of entities, accelerate the query by filtering already
@@ -263,6 +276,8 @@ def _get_states_with_session(
         query = query.join(
             most_recent_state_ids,
             States.state_id == most_recent_state_ids.c.max_state_id,
+        ).outerjoin(
+            StateAttributes, (States.attributes_id == StateAttributes.attributes_id)
         )
     else:
         # We did not get an include-list of entities, query all states in the inner
@@ -301,7 +316,9 @@ def _get_states_with_session(
         query = query.filter(~States.domain.in_(IGNORE_DOMAINS))
         if filters:
             query = filters.apply(query)
-
+        query = query.outerjoin(
+            StateAttributes, (States.attributes_id == StateAttributes.attributes_id)
+        )
     return [LazyState(row) for row in execute(query)]
 
 
@@ -315,6 +332,9 @@ def _get_single_entity_states_with_session(hass, session, utc_point_in_time, ent
         States.last_updated < bindparam("utc_point_in_time"),
         States.entity_id == bindparam("entity_id"),
     )
+    baked_query += lambda q: q.outerjoin(
+        StateAttributes, States.attributes_id == StateAttributes.attributes_id
+    )
     baked_query += lambda q: q.order_by(States.last_updated.desc())
     baked_query += lambda q: q.limit(1)
 

homeassistant/components/recorder/manifest.json

@@ -2,7 +2,7 @@
   "domain": "recorder",
   "name": "Recorder",
   "documentation": "https://www.home-assistant.io/integrations/recorder",
-  "requirements": ["sqlalchemy==1.4.32"],
+  "requirements": ["sqlalchemy==1.4.32", "fnvhash==0.1.0","lru-dict==1.1.7"],
   "codeowners": ["@home-assistant/core"],
   "quality_scale": "internal",
   "iot_class": "local_push"

homeassistant/components/recorder/migration.py

@@ -638,6 +638,9 @@ def _apply_update(instance, new_version, old_version):  # noqa: C901
                 "statistics_short_term",
                 "ix_statistics_short_term_statistic_id_start",
             )
+    elif new_version == 25:
+        _add_columns(instance, "states", ["attributes_id INTEGER"])
+        _create_index(instance, "states", "ix_states_attributes_id")
 
     else:
         raise ValueError(f"No schema migration defined for version {new_version}")