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
@@ -18,6 +19,7 @@
 from homeassistant.components.recorder import get_instance
 from homeassistant.components.recorder.models import (
     Events,
+    StateAttributes,
     States,
     process_timestamp_to_utc_isoformat,
 )
@@ -494,6 +496,7 @@ def _generate_events_query(session):
         States.entity_id,
         States.domain,
         States.attributes,
+        StateAttributes.shared_attrs,
     )
 
 
@@ -504,6 +507,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"),
     )
 
 
@@ -519,6 +523,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)
+        )
     )
 
 
@@ -534,7 +541,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):
@@ -556,6 +565,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)
+        ),
     )
 
 
@@ -709,8 +721,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
@@ -734,14 +747,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)
         return self._attributes
 
     @property
@@ -772,12 +783,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/plant/__init__.py

@@ -7,7 +7,7 @@
 import voluptuous as vol
 
 from homeassistant.components.recorder import get_instance
-from homeassistant.components.recorder.models import States
+from homeassistant.components.recorder.models import StateAttributes, States
 from homeassistant.components.recorder.util import execute, session_scope
 from homeassistant.const import (
     ATTR_TEMPERATURE,
@@ -110,11 +110,6 @@
 CONFIG_SCHEMA = vol.Schema({DOMAIN: {cv.string: PLANT_SCHEMA}}, extra=vol.ALLOW_EXTRA)
 
 
-# Flag for enabling/disabling the loading of the history from the database.
-# This feature is turned off right now as its tests are not 100% stable.
-ENABLE_LOAD_HISTORY = False
-
-
 async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     """Set up the Plant component."""
     component = EntityComponent(_LOGGER, DOMAIN, hass)
@@ -282,7 +277,7 @@ def _check_max(self, sensor_name, value, params):
 
     async def async_added_to_hass(self):
         """After being added to hass, load from history."""
-        if ENABLE_LOAD_HISTORY and "recorder" in self.hass.config.components:
+        if "recorder" in self.hass.config.components:
             # only use the database if it's configured
             await get_instance(self.hass).async_add_executor_job(
                 self._load_history_from_db
@@ -315,14 +310,24 @@ def _load_history_from_db(self):
         _LOGGER.debug("Initializing values for %s from the database", self._name)
         with session_scope(hass=self.hass) as session:
             query = (
-                session.query(States)
+                session.query(States, StateAttributes)
                 .filter(
                     (States.entity_id == entity_id.lower())
                     and (States.last_updated > start_date)
                 )
+                .outerjoin(
+                    StateAttributes,
+                    States.attributes_id == StateAttributes.attributes_id,
+                )
                 .order_by(States.last_updated.asc())
             )
-            states = execute(query, to_native=True, validate_entity_ids=False)
+            states = []
+            if results := execute(query, to_native=False, validate_entity_ids=False):
+                for state, attributes in results:
+                    native = state.to_native()
+                    if not native.attributes:
+                        native.attributes = attributes.to_native()
+                    states.append(native)
 
             for state in states:
                 # filter out all None, NaN and "unknown" states

homeassistant/components/recorder/__init__.py

@@ -14,6 +14,7 @@
 import time
 from typing import Any, TypeVar
 
+from lru import LRU  # pylint: disable=no-name-in-module
 from sqlalchemy import create_engine, event as sqlalchemy_event, exc, func, select
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.orm import scoped_session, sessionmaker
@@ -67,6 +68,7 @@
     Base,
     Events,
     RecorderRuns,
+    StateAttributes,
     States,
     StatisticsRuns,
     process_timestamp,
@@ -131,6 +133,15 @@
 # States and Events objects
 EXPIRE_AFTER_COMMITS = 120
 
+# The number of attribute ids to cache in memory
+#
+# Based on:
+# - The number of overlapping attributes
+# - How frequently states with overlapping attributes will change
+# - How much memory our low end hardware has
+STATE_ATTRIBUTES_ID_CACHE_SIZE = 2048
+
+
 DB_LOCK_TIMEOUT = 30
 DB_LOCK_QUEUE_CHECK_TIMEOUT = 1
 
@@ -541,6 +552,8 @@ def __init__(
         self._commits_without_expire = 0
         self._keepalive_count = 0
         self._old_states: dict[str, States] = {}
+        self._state_attributes_ids: LRU = LRU(STATE_ATTRIBUTES_ID_CACHE_SIZE)
+        self._pending_state_attributes: dict[str, StateAttributes] = {}
         self._pending_expunge: list[States] = []
         self.event_session = None
         self.get_session = None
@@ -964,33 +977,58 @@ 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 = 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,
                 )
+                return
+
+            dbstate.attributes = None
+            shared_attrs = dbstate_attributes.shared_attrs
+            # Matching attributes found in the pending commit
+            if pending_attributes := self._pending_state_attributes.get(shared_attrs):
+                dbstate.state_attributes = pending_attributes
+            # Matching attributes id found in the cache
+            elif attributes_id := self._state_attributes_ids.get(shared_attrs):
+                dbstate.attributes_id = attributes_id
+            # Matching attributes found in the database
+            elif (
+                attributes := self.event_session.query(StateAttributes.attributes_id)
+                .filter(StateAttributes.hash == dbstate_attributes.hash)
+                .filter(StateAttributes.shared_attrs == shared_attrs)
+                .first()
+            ):
+                dbstate.attributes_id = attributes[0]
+                self._state_attributes_ids[shared_attrs] = attributes[0]
+            # No matching attributes found, save them in the DB
+            else:
+                dbstate.state_attributes = dbstate_attributes
+                self._pending_state_attributes[shared_attrs] = dbstate_attributes
+                self.event_session.add(dbstate_attributes)
+
+            if old_state := self._old_states.pop(dbstate.entity_id, None):
+                if old_state.state_id:
+                    dbstate.old_state_id = old_state.state_id
+                else:
+                    dbstate.old_state = old_state
+            if event.data.get("new_state"):
+                self._old_states[dbstate.entity_id] = dbstate
+                self._pending_expunge.append(dbstate)
+            else:
+                dbstate.state = None
+            self.event_session.add(dbstate)
+            dbstate.event = dbevent
 
         # If they do not have a commit interval
         # than we commit right away
@@ -1042,6 +1080,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
@@ -1062,6 +1101,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

@@ -13,7 +13,12 @@
 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 +51,7 @@
     States.attributes,
     States.last_changed,
     States.last_updated,
+    StateAttributes.shared_attrs,
 ]
 
 HISTORY_BAKERY = "recorder_history_bakery"
@@ -114,6 +120,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 +168,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(
@@ -186,6 +198,9 @@ def get_last_state_changes(hass, number_of_states, entity_id):
             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()
         )
@@ -263,6 +278,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 +318,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 +334,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"