Do not purge the most recent state for an entity (#11039)

* Protect states that are the most recent states of their entity

* Also protect events

* Some documentation

* Fix SQL

homeassistant/components/recorder/purge.py

@@ -12,17 +12,44 @@ _LOGGER = logging.getLogger(__name__)
 def purge_old_data(instance, purge_days):
     """Purge events and states older than purge_days ago."""
     from .models import States, Events
+    from sqlalchemy import func
+
     purge_before = dt_util.utcnow() - timedelta(days=purge_days)
 
     with session_scope(session=instance.get_session()) as session:
+        # For each entity, the most recent state is protected from deletion
+        # s.t. we can properly restore state even if the entity has not been
+        # updated in a long time
+        protected_states = session.query(States.state_id, States.event_id,
+                                         func.max(States.last_updated)) \
+                              .group_by(States.entity_id).subquery()
+
+        protected_state_ids = session.query(States.state_id).join(
+            protected_states, States.state_id == protected_states.c.state_id)\
+            .subquery()
+
         deleted_rows = session.query(States) \
                               .filter((States.last_updated < purge_before)) \
+                              .filter(~States.state_id.in_(
+                                  protected_state_ids)) \
                               .delete(synchronize_session=False)
         _LOGGER.debug("Deleted %s states", deleted_rows)
 
+        # We also need to protect the events belonging to the protected states.
+        # Otherwise, if the SQL server has "ON DELETE CASCADE" as default, it
+        # will delete the protected state when deleting its associated
+        # event. Also, we would be producing NULLed foreign keys otherwise.
+
+        protected_event_ids = session.query(States.event_id).join(
+            protected_states, States.state_id == protected_states.c.state_id)\
+            .filter(~States.event_id is not None).subquery()
+
         deleted_rows = session.query(Events) \
-                              .filter((Events.time_fired < purge_before)) \
-                              .delete(synchronize_session=False)
+            .filter((Events.time_fired < purge_before)) \
+            .filter(~Events.event_id.in_(
+                protected_event_ids
+            )) \
+            .delete(synchronize_session=False)
         _LOGGER.debug("Deleted %s events", deleted_rows)
 
     # Execute sqlite vacuum command to free up space on disk

tests/components/recorder/test_purge.py

@@ -55,6 +55,23 @@ class TestRecorderPurge(unittest.TestCase):
                     event_id=event_id + 1000
                 ))
 
+            # if self._add_test_events was called, we added a special event
+            # that should be protected from deletion, too
+            protected_event_id = getattr(self, "_protected_event_id", 2000)
+
+            # add a state that is old but the only state of its entity and
+            # should be protected
+            session.add(States(
+                entity_id='test.rarely_updated_entity',
+                domain='sensor',
+                state='iamprotected',
+                attributes=json.dumps(attributes),
+                last_changed=five_days_ago,
+                last_updated=five_days_ago,
+                created=five_days_ago,
+                event_id=protected_event_id
+            ))
+
     def _add_test_events(self):
         """Add a few events for testing."""
         now = datetime.now()
@@ -81,19 +98,32 @@ class TestRecorderPurge(unittest.TestCase):
                     time_fired=timestamp,
                 ))
 
+            # Add an event for the protected state
+            protected_event = Events(
+                event_type='EVENT_TEST_FOR_PROTECTED',
+                event_data=json.dumps(event_data),
+                origin='LOCAL',
+                created=five_days_ago,
+                time_fired=five_days_ago,
+            )
+            session.add(protected_event)
+            session.flush()
+
+            self._protected_event_id = protected_event.event_id
+
     def test_purge_old_states(self):
         """Test deleting old states."""
         self._add_test_states()
-        # make sure we start with 5 states
+        # make sure we start with 6 states
         with session_scope(hass=self.hass) as session:
             states = session.query(States)
-            self.assertEqual(states.count(), 5)
+            self.assertEqual(states.count(), 6)
 
             # run purge_old_data()
             purge_old_data(self.hass.data[DATA_INSTANCE], 4)
 
-            # we should only have 2 states left after purging
-            self.assertEqual(states.count(), 2)
+            # we should only have 3 states left after purging
+            self.assertEqual(states.count(), 3)
 
     def test_purge_old_events(self):
         """Test deleting old events."""
@@ -102,7 +132,7 @@ class TestRecorderPurge(unittest.TestCase):
         with session_scope(hass=self.hass) as session:
             events = session.query(Events).filter(
                 Events.event_type.like("EVENT_TEST%"))
-            self.assertEqual(events.count(), 5)
+            self.assertEqual(events.count(), 6)
 
             # run purge_old_data()
             purge_old_data(self.hass.data[DATA_INSTANCE], 4)
@@ -113,17 +143,17 @@ class TestRecorderPurge(unittest.TestCase):
     def test_purge_method(self):
         """Test purge method."""
         service_data = {'keep_days': 4}
-        self._add_test_states()
         self._add_test_events()
+        self._add_test_states()
 
-        # make sure we start with 5 states
+        # make sure we start with 6 states
         with session_scope(hass=self.hass) as session:
             states = session.query(States)
-            self.assertEqual(states.count(), 5)
+            self.assertEqual(states.count(), 6)
 
             events = session.query(Events).filter(
                 Events.event_type.like("EVENT_TEST%"))
-            self.assertEqual(events.count(), 5)
+            self.assertEqual(events.count(), 6)
 
             self.hass.data[DATA_INSTANCE].block_till_done()
 
@@ -134,11 +164,9 @@ class TestRecorderPurge(unittest.TestCase):
             # Small wait for recorder thread
             sleep(0.1)
 
-            # we should only have 2 states left after purging
-            self.assertEqual(states.count(), 5)
-
-            # now we should only have 3 events left
-            self.assertEqual(events.count(), 5)
+            # we should still have everything from before
+            self.assertEqual(states.count(), 6)
+            self.assertEqual(events.count(), 6)
 
             # run purge method - correct service data
             self.hass.services.call('recorder', 'purge',
@@ -148,8 +176,18 @@ class TestRecorderPurge(unittest.TestCase):
             # Small wait for recorder thread
             sleep(0.1)
 
-            # we should only have 2 states left after purging
-            self.assertEqual(states.count(), 2)
+            # we should only have 3 states left after purging
+            self.assertEqual(states.count(), 3)
 
-            # now we should only have 3 events left
-            self.assertEqual(events.count(), 3)
+            # the protected state is among them
+            self.assertTrue('iamprotected' in (
+                state.state for state in states))
+
+            # now we should only have 4 events left
+            self.assertEqual(events.count(), 4)
+
+            # and the protected event is among them
+            self.assertTrue('EVENT_TEST_FOR_PROTECTED' in (
+                event.event_type for event in events.all()))
+            self.assertFalse('EVENT_TEST_PURGE' in (
+                event.event_type for event in events.all()))