Repository: home-assistant/core
Pull request: Ensure purge can cleanup old format detached states in the database

## Proposed change
<!--
  Describe the big picture of your changes here to communicate to the
  maintainers why we should accept this pull request. If it fixes a bug
  or resolves a feature request, be sure to link to that issue in the
  additional information section.
-->
If there were still rows in the states table that had an event_id but the event had been deleted we would never be able to switch over the to purging now format rows since these rows were detached from the event table and would previously never been purged

fixes #92120


## Type of change
<!--
  What type of change does your PR introduce to Home Assistant?
  NOTE: Please, check only 1! box!
  If your PR requires multiple boxes to be checked, you'll most likely need to
  split it into multiple PRs. This makes things easier and faster to code review.
-->

- [ ] Dependency upgrade
- [x] Bugfix (non-breaking change which fixes an issue)
- [ ] New integration (thank you!)
- [ ] New feature (which adds functionality to an existing integration)
- [ ] Deprecation (breaking change to happen in the future)
- [ ] Breaking change (fix/feature causing existing functionality to break)
- [ ] Code quality improvements to existing code or addition of tests

## Additional information
<!--
  Details are important, and help maintainers processing your PR.
  Please be sure to fill out additional details, if applicable.
-->

- This PR fixes or closes issue: fixes #
- This PR is related to issue: 
- Link to documentation pull request: 

## Checklist
<!--
  Put an `x` in the boxes that apply. You can also fill these out after
  creating the PR. If you're unsure about any of them, don't hesitate to ask.
  We're here to help! This is simply a reminder of what we are going to look
  for before merging your code.
-->

- [x] The code change is tested and works locally.
- [ ] Local tests pass. **Your PR cannot be merged unless tests pass**
- [ ] There is no commented out code in this PR.
- [ ] I have followed the [development checklist][dev-checklist]
- [ ] I have followed the [perfect PR recommendations][perfect-pr]
- [ ] The code has been formatted using Black (`black --fast homeassistant tests`)
- [x] Tests have been added to verify that the new code works.

If user exposed functionality or configuration variables are added/changed:

- [ ] Documentation added/updated for [www.home-assistant.io][docs-repository]

If the code communicates with devices, web services, or third-party tools:

- [ ] The [manifest file][manifest-docs] has all fields filled out correctly.  
      Updated and included derived files by running: `python3 -m script.hassfest`.
- [ ] New or updated dependencies have been added to `requirements_all.txt`.  
      Updated by running `python3 -m script.gen_requirements_all`.
- [ ] For the updated dependencies - a link to the changelog, or at minimum a diff between library versions is added to the PR description.
- [ ] Untested files have been added to `.coveragerc`.

<!--
  This project is very active and we have a high turnover of pull requests.

  Unfortunately, the number of incoming pull requests is higher than what our
  reviewers can review and merge so there is a long backlog of pull requests
  waiting for review. You can help here!
  
  By reviewing another pull request, you will help raise the code quality of
  that pull request and the final review will be faster. This way the general
  pace of pull request reviews will go up and your wait time will go down.
  
  When picking a pull request to review, try to choose one that hasn't yet
  been reviewed.

  Thanks for helping out!
-->

To help with the load of incoming pull requests:

- [ ] I have reviewed two other [open pull requests][prs] in this repository.

[prs]: https://github.com/home-assistant/core/pulls?q=is%3Aopen+is%3Apr+-author%3A%40me+-draft%3Atrue+-label%3Awaiting-for-upstream+sort%3Acreated-desc+review%3Anone+-status%3Afailure

<!--
  Thank you for contributing <3

  Below, some useful links you could explore:
-->
[dev-checklist]: https://developers.home-assistant.io/docs/en/development_checklist.html
[manifest-docs]: https://developers.home-assistant.io/docs/en/creating_integration_manifest.html
[quality-scale]: https://developers.home-assistant.io/docs/en/next/integration_quality_scale_index.html
[docs-repository]: https://github.com/home-assistant/home-assistant.io
[perfect-pr]: https://developers.home-assistant.io/docs/review-process/#creating-the-perfect-pr


Problem statement:
recorder purge not purging with "Purge running in legacy format as there are states with event_id remaining"
### The problem

I have a sqlite db that is not purging neither automatically ( in the night routine ) nor by calling directly the service from developers tool.
At the moment my states tables contains +3mln rows starting from 2022-11-26ish and keep growing.
In configuration.yaml I have the following:
``` 
recorder:
  purge_keep_days: 10
  commit_interval: 300
  auto_repack: true
  auto_purge: true
```
but also calling the service with ( eventually )  more days to keep I always end up with this message :+1: 
```
istant
2023-04-27 15:37:05.988 DEBUG (Recorder) [homeassistant.components.recorder.purge] Purging states and events before target 2023-03-28 13:37:05+00:00
2023-04-27 15:37:05.995 DEBUG (Recorder) [homeassistant.components.recorder.purge] Purge running in legacy format as there are states with event_id remaining
2023-04-27 15:37:06.006 DEBUG (Recorder) [homeassistant.components.recorder.purge] Selected 0 event ids to remove
2023-04-27 15:37:06.018 DEBUG (Recorder) [homeassistant.components.recorder.purge] Selected 0 statistic runs to remove
2023-04-27 15:37:06.027 DEBUG (Recorder) [homeassistant.components.recorder.purge] Selected 0 short term statistics to remove
2023-04-27 15:37:06.807 DEBUG (Recorder) [homeassistant.components.recorder.purge] Deleted <sqlalchemy.engine.cursor.CursorResult object at 0x9e1139e8> recorder_runs
```

Obviously I would rather not delete the db and start again as I have a bunch of long term statistics that I would like to keep.

### What version of Home Assistant Core has the issue?

core-2023.4.2

### What was the last working version of Home Assistant Core?

_No response_

### What type of installation are you running?

Home Assistant Container

### Integration causing the issue

_No response_

### Link to integration documentation on our website

_No response_

### Diagnostics information

_No response_

### Example YAML snippet

_No response_

### Anything in the logs that might be useful for us?

_No response_

### Additional information

_No response_

Hints from issue discussion:
Hey there @home-assistant/core, mind taking a look at this issue as it has been labeled with an integration (`recorder`) you are listed as a [code owner](https://github.com/home-assistant/core/blob/dev/CODEOWNERS#L976) for? Thanks!

<details>
<summary>Code owner commands</summary>


Code owners of `recorder` can trigger bot actions by commenting:

- `@home-assistant close` Closes the issue.
- `@home-assistant rename Awesome new title` Renames the issue.
- `@home-assistant reopen` Reopen the issue.
- `@home-assistant unassign recorder` Removes the current integration label and assignees on the issue, add the integration domain after the command.

</details>


<sub><sup>(message by CodeOwnersMention)</sup></sub>

---

[recorder documentation](https://www.home-assistant.io/integrations/recorder)
[recorder source](https://github.com/home-assistant/core/tree/dev/homeassistant/components/recorder)
<sub><sup>(message by IssueLinks)</sup></sub>
I'll need a copy of the database file to see what has gone wrong.

My dropbox is nick@koston.org or gdrive is bdraco@gmail.com

Thanks
I should have sent to your gdrive.
Let me know if you can get it, please.
Never tried to share via gdrive.
Thanks ! 
Great. I'm importing it now and will test shortly
It looks like you have 5 detached states that are not linked to any event in the database from `2022-01-27` that are causing the issue.

I will see if I can come up with a way to detect and remove them safely
 I put together a solution for this.  I'll submit it shortly

Commit message: cover

Patch to review:
diff --git a/homeassistant/components/recorder/purge.py b/homeassistant/components/recorder/purge.py
index 662be41b1c8e5..95013de125dcb 100644
--- a/homeassistant/components/recorder/purge.py
+++ b/homeassistant/components/recorder/purge.py
@@ -34,6 +34,7 @@
     find_event_types_to_purge,
     find_events_to_purge,
     find_latest_statistics_runs_run_id,
+    find_legacy_detached_states_and_attributes_to_purge,
     find_legacy_event_state_and_attributes_and_data_ids_to_purge,
     find_legacy_row,
     find_short_term_statistics_to_purge,
@@ -146,7 +147,28 @@ def _purge_legacy_format(
     _purge_unused_attributes_ids(instance, session, attributes_ids)
     _purge_event_ids(session, event_ids)
     _purge_unused_data_ids(instance, session, data_ids)
-    return bool(event_ids or state_ids or attributes_ids or data_ids)
+
+    # The database may still have some rows that have an event_id but are not
+    # linked to any event. These rows are not linked to any event because the
+    # event was deleted. We need to purge these rows as well or we will never
+    # switch to the new format which will prevent us from purging any events
+    # that happened after the detached states.
+    (
+        detached_state_ids,
+        detached_attributes_ids,
+    ) = _select_legacy_detached_state_and_attributes_and_data_ids_to_purge(
+        session, purge_before
+    )
+    _purge_state_ids(instance, session, detached_state_ids)
+    _purge_unused_attributes_ids(instance, session, detached_attributes_ids)
+    return bool(
+        event_ids
+        or state_ids
+        or attributes_ids
+        or data_ids
+        or detached_state_ids
+        or detached_attributes_ids
+    )
 
 
 def _purge_states_and_attributes_ids(
@@ -412,6 +434,31 @@ def _select_short_term_statistics_to_purge(
     return [statistic.id for statistic in statistics]
 
 
+def _select_legacy_detached_state_and_attributes_and_data_ids_to_purge(
+    session: Session, purge_before: datetime
+) -> tuple[set[int], set[int]]:
+    """Return a list of state, and attribute ids to purge.
+
+    We do not link these anymore since state_change events
+    do not exist in the events table anymore, however we
+    still need to be able to purge them.
+    """
+    states = session.execute(
+        find_legacy_detached_states_and_attributes_to_purge(
+            dt_util.utc_to_timestamp(purge_before)
+        )
+    ).all()
+    _LOGGER.debug("Selected %s state ids to remove", len(states))
+    state_ids = set()
+    attributes_ids = set()
+    for state in states:
+        if state_id := state.state_id:
+            state_ids.add(state_id)
+        if attributes_id := state.attributes_id:
+            attributes_ids.add(attributes_id)
+    return state_ids, attributes_ids
+
+
 def _select_legacy_event_state_and_attributes_and_data_ids_to_purge(
     session: Session, purge_before: datetime
 ) -> tuple[set[int], set[int], set[int], set[int]]:
@@ -433,12 +480,12 @@ def _select_legacy_event_state_and_attributes_and_data_ids_to_purge(
     data_ids = set()
     for event in events:
         event_ids.add(event.event_id)
-        if event.state_id:
-            state_ids.add(event.state_id)
-        if event.attributes_id:
-            attributes_ids.add(event.attributes_id)
-        if event.data_id:
-            data_ids.add(event.data_id)
+        if state_id := event.state_id:
+            state_ids.add(state_id)
+        if attributes_id := event.attributes_id:
+            attributes_ids.add(attributes_id)
+        if data_id := event.data_id:
+            data_ids.add(data_id)
     return event_ids, state_ids, attributes_ids, data_ids
 
 
diff --git a/homeassistant/components/recorder/queries.py b/homeassistant/components/recorder/queries.py
index f8a1b769d87b8..49f66fdcd6827 100644
--- a/homeassistant/components/recorder/queries.py
+++ b/homeassistant/components/recorder/queries.py
@@ -678,6 +678,22 @@ def find_legacy_event_state_and_attributes_and_data_ids_to_purge(
     )
 
 
+def find_legacy_detached_states_and_attributes_to_purge(
+    purge_before: float,
+) -> StatementLambdaElement:
+    """Find states rows with event_id set but not linked event_id in Events."""
+    return lambda_stmt(
+        lambda: select(States.state_id, States.attributes_id)
+        .outerjoin(Events, States.event_id == Events.event_id)
+        .filter(States.event_id.isnot(None))
+        .filter(
+            (States.last_updated_ts < purge_before) | States.last_updated_ts.is_(None)
+        )
+        .filter(Events.event_id.is_(None))
+        .limit(SQLITE_MAX_BIND_VARS)
+    )
+
+
 def find_legacy_row() -> StatementLambdaElement:
     """Check if there are still states in the table with an event_id."""
     # https://github.com/sqlalchemy/sqlalchemy/issues/9189
diff --git a/tests/components/recorder/test_purge_v32_schema.py b/tests/components/recorder/test_purge_v32_schema.py
index 433ff01eb91fe..9bba55222de45 100644
--- a/tests/components/recorder/test_purge_v32_schema.py
+++ b/tests/components/recorder/test_purge_v32_schema.py
@@ -8,6 +8,7 @@
 
 from freezegun import freeze_time
 import pytest
+from sqlalchemy import text, update
 from sqlalchemy.exc import DatabaseError, OperationalError
 from sqlalchemy.orm.session import Session
 
@@ -1000,7 +1001,7 @@ async def test_purge_many_old_events(
 async def test_purge_can_mix_legacy_and_new_format(
     async_setup_recorder_instance: RecorderInstanceGenerator, hass: HomeAssistant
 ) -> None:
-    """Test purging with legacy a new events."""
+    """Test purging with legacy and new events."""
     instance = await async_setup_recorder_instance(hass)
     await _async_attach_db_engine(hass)
 
@@ -1018,6 +1019,7 @@ def _recreate_legacy_events_index():
 
     utcnow = dt_util.utcnow()
     eleven_days_ago = utcnow - timedelta(days=11)
+
     with session_scope(hass=hass) as session:
         broken_state_no_time = States(
             event_id=None,
@@ -1104,6 +1106,150 @@ def _recreate_legacy_events_index():
         assert states_without_event_id.count() == 1
 
 
+async def test_purge_can_mix_legacy_and_new_format_with_detached_state(
+    async_setup_recorder_instance: RecorderInstanceGenerator,
+    hass: HomeAssistant,
+    recorder_db_url: str,
+) -> None:
+    """Test purging with legacy a new events with a detached state."""
+    if recorder_db_url.startswith(("mysql://", "postgresql://")):
+        return pytest.skip("This tests disables foreign key checks on SQLite")
+
+    instance = await async_setup_recorder_instance(hass)
+    await _async_attach_db_engine(hass)
+
+    await async_wait_recording_done(hass)
+    # New databases are no longer created with the legacy events index
+    assert instance.use_legacy_events_index is False
+
+    def _recreate_legacy_events_index():
+        """Recreate the legacy events index since its no longer created on new instances."""
+        migration._create_index(instance.get_session, "states", "ix_states_event_id")
+        instance.use_legacy_events_index = True
+
+    await instance.async_add_executor_job(_recreate_legacy_events_index)
+    assert instance.use_legacy_events_index is True
+
+    with session_scope(hass=hass) as session:
+        session.execute(text("PRAGMA foreign_keys = OFF"))
+
+    utcnow = dt_util.utcnow()
+    eleven_days_ago = utcnow - timedelta(days=11)
+
+    with session_scope(hass=hass) as session:
+        broken_state_no_time = States(
+            event_id=None,
+            entity_id="orphened.state",
+            last_updated_ts=None,
+            last_changed_ts=None,
+        )
+        session.add(broken_state_no_time)
+        detached_state_deleted_event_id = States(
+            event_id=99999999999,
+            entity_id="event.deleted",
+            last_updated_ts=1,
+            last_changed_ts=None,
+        )
+        session.add(detached_state_deleted_event_id)
+        detached_state_deleted_event_id.last_changed = None
+        detached_state_deleted_event_id.last_changed_ts = None
+        detached_state_deleted_event_id.last_updated = None
+        detached_state_deleted_event_id = States(
+            event_id=99999999999,
+            entity_id="event.deleted.no_time",
+            last_updated_ts=None,
+            last_changed_ts=None,
+        )
+        detached_state_deleted_event_id.last_changed = None
+        detached_state_deleted_event_id.last_changed_ts = None
+        detached_state_deleted_event_id.last_updated = None
+        detached_state_deleted_event_id.last_updated_ts = None
+        session.add(detached_state_deleted_event_id)
+        start_id = 50000
+        for event_id in range(start_id, start_id + 50):
+            _add_state_and_state_changed_event(
+                session,
+                "sensor.excluded",
+                "purgeme",
+                eleven_days_ago,
+                event_id,
+            )
+    with session_scope(hass=hass) as session:
+        session.execute(
+            update(States)
+            .where(States.entity_id == "event.deleted.no_time")
+            .values(last_updated_ts=None)
+        )
+
+    await _add_test_events(hass, 50)
+    await _add_events_with_event_data(hass, 50)
+    with session_scope(hass=hass) as session:
+        for _ in range(50):
+            _add_state_without_event_linkage(
+                session, "switch.random", "on", eleven_days_ago
+            )
+        states_with_event_id = session.query(States).filter(
+            States.event_id.is_not(None)
+        )
+        states_without_event_id = session.query(States).filter(
+            States.event_id.is_(None)
+        )
+
+        assert states_with_event_id.count() == 52
+        assert states_without_event_id.count() == 51
+
+        purge_before = dt_util.utcnow() - timedelta(days=4)
+        finished = purge_old_data(
+            instance,
+            purge_before,
+            repack=False,
+        )
+        assert not finished
+        assert states_with_event_id.count() == 0
+        assert states_without_event_id.count() == 51
+        # At this point all the legacy states are gone
+        # and we switch methods
+        purge_before = dt_util.utcnow() - timedelta(days=4)
+        finished = purge_old_data(
+            instance,
+            purge_before,
+            repack=False,
+            events_batch_size=1,
+            states_batch_size=1,
+        )
+        # Since we only allow one iteration, we won't
+        # check if we are finished this loop similar
+        # to the legacy method
+        assert not finished
+        assert states_with_event_id.count() == 0
+        assert states_without_event_id.count() == 1
+        finished = purge_old_data(
+            instance,
+            purge_before,
+            repack=False,
+            events_batch_size=100,
+            states_batch_size=100,
+        )
+        assert finished
+        assert states_with_event_id.count() == 0
+        assert states_without_event_id.count() == 1
+        _add_state_without_event_linkage(
+            session, "switch.random", "on", eleven_days_ago
+        )
+        assert states_with_event_id.count() == 0
+        assert states_without_event_id.count() == 2
+        finished = purge_old_data(
+            instance,
+            purge_before,
+            repack=False,
+        )
+        assert finished
+        # The broken state without a timestamp
+        # does not prevent future purges. Its ignored.
+        assert states_with_event_id.count() == 0
+        assert states_without_event_id.count() == 1
+
+
 async def test_purge_entities_keep_days(
     async_setup_recorder_instance: RecorderInstanceGenerator,
     hass: HomeAssistant,


Code review comments:
diff_hunk: @@ -1104,6 +1106,150 @@ def _recreate_legacy_events_index():
         assert states_without_event_id.count() == 1
 
 
+async def test_purge_can_mix_legacy_and_new_format_with_detached_state(
+    async_setup_recorder_instance: RecorderInstanceGenerator,
+    hass: HomeAssistant,
+    recorder_db_url: str,
+) -> None:
+    """Test purging with legacy a new events with a detached state."""
path: tests/components/recorder/test_purge_v32_schema.py
text: @author:
```suggestion
    """Test purging with legacy and new events with a detached state."""
```