Repository: home-assistant/core
Pull request: Cache the latest short term stat id for each metadata_id on each run
## 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.
-->
We currently have to do a algorithmically complex query to find the latest short term stat id every five minutes when stats are run which results in a full scan with PostgreSQL. To avoid this we now cache the short term statistics `id` for each `metadata_id` in memory between runs.

During the first short term stats run after restart, we populate the cache with a query per metadata_id since the combined query was significantly slower.


## 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
- [ ] 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)
- [x] 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 #94435
- 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/development_checklist/
[manifest-docs]: https://developers.home-assistant.io/docs/creating_integration_manifest/
[quality-scale]: https://developers.home-assistant.io/docs/integration_quality_scale_index/
[docs-repository]: https://github.com/home-assistant/home-assistant.io
[perfect-pr]: https://developers.home-assistant.io/docs/review-process/#creating-the-perfect-pr

TODO: 
- [x] consider if we need to invalidate the data if they upgrade, downgrade, upgrade and don't rollback the database (this isn't supported but people will do it, but it might be too expensive or complex to handle this). This is handled by removing the latest table at startup. This means a manual rebuild of it the first time we run 5 minute stats, but we can get rid of this after a few versions.  The manual rebuild isn't that much more expensive than the existing query so its not a significant consideration.


Problem statement:
Slow running queries in HA 2023.6.1 with Postgres
### The problem

Postgres 13.9 with a large database

After updating from 2023.5.3 to 2023.6.1, I noticed the following query running every 5 minutes.  It takes about 90 seconds to complete and spikes the CPU during that time.

```
SELECT statistics_short_term.metadata_id, statistics_short_term.start_ts, statistics_short_term.mean, statistics_short_term.min, statistics_short_term.max, statistics_short_term.last_reset_ts, statistics_short_term.state, statistics_short_term.sum 
FROM statistics_short_term JOIN (
  SELECT statistics_short_term.metadata_id AS metadata_id, max(statistics_short_term.start_ts) AS start_max 
  FROM statistics_short_term 
  WHERE statistics_short_term.metadata_id IN (32, 206, 46, 200, 127, 125, 129, 117, 112, 80, 203, 111, 132, 115, 16, 143, 113, 34, 156, 41, 67, 130, 109, 68, 108, 66, 103, 21, 96, 40, 153, 114, 49, 42, 144, 28, 107, 151, 116, 33, 75, 48, 102, 73, 100, 138, 72, 98, 81, 133, 202, 36, 74, 29, 207, 145, 128, 159, 70, 94, 131, 142, 25, 69, 95, 99, 184, 47, 30, 15, 154, 126, 35, 157, 26, 86, 158, 60, 38, 71, 152, 148, 105, 22, 97, 155, 27, 201, 186, 196) 
  GROUP BY statistics_short_term.metadata_id
) AS anon_1 ON statistics_short_term.metadata_id = anon_1.metadata_id AND statistics_short_term.start_ts = anon_1.start_max
```

After trying to break down the query, I noticed the following simple query can take 1-2 seconds to run:

```
SELECT statistics_short_term.metadata_id AS metadata_id, 
max(statistics_short_term.start_ts) AS start_max 
FROM statistics_short_term 
WHERE statistics_short_term.metadata_id IN (32) 
GROUP BY statistics_short_term.metadata_id
```

Explain analyze:
```
"GroupAggregate  (cost=0.56..129573.43 rows=145 width=12) (actual time=962.568..962.570 rows=1 loops=1)"
"  Group Key: metadata_id"
"  ->  Index Only Scan using ix_statistics_short_term_statistic_id_start_ts on statistics_short_term  (cost=0.56..128757.89 rows=162818 width=12) (actual time=2.218..893.909 rows=170531 loops=1)"
"        Index Cond: (metadata_id = 32)"
"        Heap Fetches: 41957"
"Planning Time: 0.235 ms"
"JIT:"
"  Functions: 4"
"  Options: Inlining false, Optimization false, Expressions true, Deforming true"
"  Timing: Generation 1.275 ms, Inlining 0.000 ms, Optimization 0.586 ms, Emission 14.579 ms, Total 16.440 ms"
"Execution Time: 964.048 ms"
```

Modifying the query to the following runs quickly:

```
SELECT statistics_short_term.metadata_id AS metadata_id, 
statistics_short_term.start_ts AS start_max 
FROM statistics_short_term 
WHERE statistics_short_term.metadata_id = 32
ORDER BY statistics_short_term.start_ts DESC
LIMIT 1
```

Explain analyze:
```
"Limit  (cost=0.56..1.35 rows=1 width=12) (actual time=0.303..0.304 rows=1 loops=1)"
"  ->  Index Only Scan Backward using ix_statistics_short_term_statistic_id_start_ts on statistics_short_term  (cost=0.56..128757.89 rows=162818 width=12) (actual time=0.301..0.301 rows=1 loops=1)"
"        Index Cond: (metadata_id = 32)"
"        Heap Fetches: 1"
"Planning Time: 0.263 ms"
"Execution Time: 0.344 ms"
```

I downgraded back to 2023.5.3 and noticed that the same query is running, but with a smaller list of metadata_ids, so it completes in about 15 seconds, which is why I never noticed it before.

Also, possibly unrelated, but I noticed this query running during startup that was also taking a long time (about a minute).  It returns no rows.

```
SELECT events.event_id 
FROM events 
WHERE events.context_id_bin IS NULL 
 LIMIT 1
```

Explain analyze:
```
"Limit  (cost=0.00..0.88 rows=1 width=4) (actual time=63302.874..63302.878 rows=0 loops=1)"
"  ->  Seq Scan on events  (cost=0.00..455507.73 rows=519938 width=4) (actual time=63124.419..63124.420 rows=0 loops=1)"
"        Filter: (context_id_bin IS NULL)"
"        Rows Removed by Filter: 12031207"
"Planning Time: 0.354 ms"
"Execution Time: 63303.022 ms"
```

`ix_events_context_id_bin` is present but doesn't appear to be used

I'm not that knowledgable about how indexes work in Postgres but it seems like these queries are not using them.

### What version of Home Assistant Core has the issue?

core-2023.6.1

### What was the last working version of Home Assistant Core?

_No response_

### What type of installation are you running?

Home Assistant Core

### 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#L1001) 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 believe the `events` query is from `has_events_context_ids_to_migrate()`.  The following uses the index and runs a lot faster:

```
SELECT COUNT(*)
FROM events 
WHERE events.context_id_bin IS NULL
```

```
"Finalize Aggregate  (cost=150963.77..150963.78 rows=1 width=8) (actual time=248.574..267.750 rows=1 loops=1)"
"  ->  Gather  (cost=150963.56..150963.77 rows=2 width=8) (actual time=52.742..267.695 rows=3 loops=1)"
"        Workers Planned: 2"
"        Workers Launched: 2"
"        ->  Partial Aggregate  (cost=149963.56..149963.57 rows=1 width=8) (actual time=24.240..24.242 rows=1 loops=3)"
"              ->  Parallel Index Only Scan using ix_events_context_id_bin on events  (cost=0.56..149421.95 rows=216641 width=0) (actual time=5.819..5.820 rows=0 loops=3)"
"                    Index Cond: (context_id_bin IS NULL)"
"                    Heap Fetches: 0"
"Planning Time: 23.840 ms"
"JIT:"
"  Functions: 11"
"  Options: Inlining false, Optimization false, Expressions true, Deforming true"
"  Timing: Generation 4.161 ms, Inlining 0.000 ms, Optimization 1.747 ms, Emission 52.484 ms, Total 58.392 ms"
"Execution Time: 452.639 ms"
```
Using [this guide](https://www.endpointdev.com/blog/2020/06/postgresql-improve-group-by-max-performance/) I was able to make the statistics query very fast (on the order of ms).  Here's the subquery:

```
WITH RECURSIVE t AS (
    (SELECT metadata_id, start_ts FROM statistics_short_term ORDER BY metadata_id DESC, start_ts DESC LIMIT 1)
    UNION ALL
    SELECT s.metadata_id, s.start_ts FROM t,
    LATERAL (
        SELECT metadata_id, start_ts
        FROM statistics_short_term
        WHERE (statistics_short_term.metadata_id) < (t.metadata_id)
        ORDER BY metadata_id DESC, start_ts DESC LIMIT 1
    ) s
) SELECT metadata_id, start_ts AS start_max FROM t WHERE metadata_id IN (32, 206, 46, 200, 127, 125, 129, 117, 112, 80, 203, 111, 132, 115, 16, 143, 113, 34, 156, 41, 67, 130, 109, 68, 108, 66, 103, 21, 96, 40, 153, 114, 49, 42, 144, 28, 107, 151, 116, 33, 75, 48, 102, 73, 100, 138, 72, 98, 81, 133, 202, 36, 74, 29, 207, 145, 128, 159, 70, 94, 131, 142, 25, 69, 95, 99, 184, 47, 30, 15, 154, 126, 35, 157, 26, 86, 158, 60, 38, 71, 152, 148, 105, 22, 97, 155, 27, 201, 186, 196)
```

and the full thing:

```
SELECT statistics_short_term.metadata_id, statistics_short_term.start_ts, statistics_short_term.mean, statistics_short_term.min, statistics_short_term.max, statistics_short_term.last_reset_ts, statistics_short_term.state, statistics_short_term.sum 
FROM statistics_short_term JOIN (
WITH RECURSIVE t AS (
    (SELECT metadata_id, start_ts FROM statistics_short_term ORDER BY metadata_id DESC, start_ts DESC LIMIT 1)
    UNION ALL
    SELECT s.metadata_id, s.start_ts FROM t,
    LATERAL (
        SELECT metadata_id, start_ts
        FROM statistics_short_term
        WHERE (statistics_short_term.metadata_id) < (t.metadata_id)
        ORDER BY metadata_id DESC, start_ts DESC LIMIT 1
    ) s
) SELECT metadata_id, start_ts AS start_max FROM t WHERE metadata_id IN (32, 206, 46, 200, 127, 125, 129, 117, 112, 80, 203, 111, 132, 115, 16, 143, 113, 34, 156, 41, 67, 130, 109, 68, 108, 66, 103, 21, 96, 40, 153, 114, 49, 42, 144, 28, 107, 151, 116, 33, 75, 48, 102, 73, 100, 138, 72, 98, 81, 133, 202, 36, 74, 29, 207, 145, 128, 159, 70, 94, 131, 142, 25, 69, 95, 99, 184, 47, 30, 15, 154, 126, 35, 157, 26, 86, 158, 60, 38, 71, 152, 148, 105, 22, 97, 155, 27, 201, 186, 196)
) AS anon_1 ON statistics_short_term.metadata_id = anon_1.metadata_id AND statistics_short_term.start_ts = anon_1.start_max
```

```
"Nested Loop  (cost=73.50..859.42 rows=79 width=60) (actual time=0.403..6.644 rows=90 loops=1)"
"  ->  CTE Scan on t  (cost=72.94..86.32 rows=90 width=12) (actual time=0.375..5.359 rows=90 loops=1)"
"        Filter: (metadata_id = ANY ('{32,206,46,200,127,125,129,117,112,80,203,111,132,115,16,143,113,34,156,41,67,130,109,68,108,66,103,21,96,40,153,114,49,42,144,28,107,151,116,33,75,48,102,73,100,138,72,98,81,133,202,36,74,29,207,145,128,159,70,94,131,142,25,69,95,99,184,47,30,15,154,126,35,157,26,86,158,60,38,71,152,148,105,22,97,155,27,201,186,196}'::integer[]))"
"        Rows Removed by Filter: 90"
"        CTE t"
"          ->  Recursive Union  (cost=0.56..72.94 rows=101 width=12) (actual time=0.061..4.894 rows=180 loops=1)"
"                ->  Limit  (cost=0.56..0.62 rows=1 width=12) (actual time=0.056..0.058 rows=1 loops=1)"
"                      ->  Index Only Scan Backward using ix_statistics_short_term_statistic_id_start_ts on statistics_short_term statistics_short_term_1  (cost=0.56..714589.82 rows=11252060 width=12) (actual time=0.053..0.053 rows=1 loops=1)"
"                            Heap Fetches: 1"
"                ->  Nested Loop  (cost=0.56..7.03 rows=10 width=12) (actual time=0.024..0.025 rows=1 loops=180)"
"                      ->  WorkTable Scan on t t_1  (cost=0.00..0.20 rows=10 width=4) (actual time=0.000..0.000 rows=1 loops=180)"
"                      ->  Limit  (cost=0.56..0.66 rows=1 width=12) (actual time=0.023..0.023 rows=1 loops=180)"
"                            ->  Index Only Scan Backward using ix_statistics_short_term_statistic_id_start_ts on statistics_short_term statistics_short_term_2  (cost=0.56..386215.36 rows=3750687 width=12) (actual time=0.022..0.022 rows=1 loops=180)"
"                                  Index Cond: (metadata_id < t_1.metadata_id)"
"                                  Heap Fetches: 130"
"  ->  Index Scan using ix_statistics_short_term_statistic_id_start_ts_desc on statistics_short_term  (cost=0.56..8.58 rows=1 width=60) (actual time=0.012..0.012 rows=1 loops=90)"
"        Index Cond: ((metadata_id = t.metadata_id) AND (start_ts = t.start_ts))"
"Planning Time: 1.221 ms"
"Execution Time: 6.781 ms"
```

I'm not sure if this is feasible to implement in Home Assistant, however.
I had that same problem, after update from 2023.3 to 2023.6.3 I had to revert back. CPU usage above limits, power consumption much higher, my SSD used all the time.
I use postgres 15.3.

Here is SQL that execute 5-9s every time
```
SELECT statistics_short_term.metadata_id, statistics_short_term.start_ts, statistics_short_term.mean, statistics_short_term.min, statistics_short_term.max, statistics_short_term.last_reset_ts, statistics_short_term.state, statistics_short_term.sum
        FROM statistics_short_term JOIN (SELECT statistics_short_term.metadata_id AS metadata_id, max(statistics_short_term.start_ts) AS start_max
        FROM statistics_short_term
        WHERE statistics_short_term.metadata_id IN (137, 130, 228, 208, 10, 134, 251, 248, 54, 196, 203, 5, 47, 117, 133, 174, 229, 182, 28, 62, 16, 210, 227, 241, 79, 36, 39, 181, 51, 9, 167, 142, 17, 222, 247, 109, 42, 77, 206, 144, 70, 146, 38, 25, 198, 233, 150, 149, 113, 30, 207, 22, 60, 225, 50, 250, 172, 56, 235, 148, 33, 183, 12, 143, 151, 26, 118, 231, 2, 13, 230, 115, 140, 23, 78, 252, 139, 53, 213, 136, 234, 132, 219, 108, 145, 170, 200, 31, 141, 64, 20, 211, 122, 1, 37, 8, 19, 215, 116, 138, 49, 221, 128, 66, 199, 110, 63, 249, 52, 41, 35, 224, 156, 59, 57, 165, 253, 4, 11, 68, 202, 204, 58, 46, 3, 205, 7, 107, 239, 218, 216, 147, 65, 186, 232, 43, 111, 45, 223, 6, 194, 201, 48, 135, 15, 195, 212, 34, 155, 27, 21, 67, 14, 29, 44, 55, 238, 32, 153, 152, 184, 214, 24, 220, 197, 209, 69, 217, 18, 112, 40, 114, 237, 240, 80, 161, 226) GROUP BY statistics_short_term.metadata_id) AS anon_1 ON statistics_short_term.metadata_id = anon_1.metadata_id AND statistics_short_term.start_ts = anon_1.start_max;
```

and it's "explain analyze":
```
 Nested Loop  (cost=1001.15..347231.35 rows=231 width=60) (actual time=5474.913..5533.213 rows=177 loops=1)
   ->  Finalize GroupAggregate  (cost=1000.59..346676.55 rows=214 width=12) (actual time=5474.858..5531.182 rows=177 loops=1)
         Group Key: statistics_short_term_1.metadata_id
         ->  Gather Merge  (cost=1000.59..346672.27 rows=428 width=12) (actual time=5474.783..5530.980 rows=354 loops=1)
               Workers Planned: 2
               Workers Launched: 2
               ->  Partial GroupAggregate  (cost=0.56..345622.85 rows=214 width=12) (actual time=86.510..3579.100 rows=118 loops=3)
                     Group Key: statistics_short_term_1.metadata_id
                     ->  Parallel Index Only Scan using ix_statistics_short_term_statistic_id_start_ts on statistics_short_term statistics_short_term_1  (cost=0.56..302971.97 rows=8529748 width=12) (actual time=0.098..2445.874 rows=6993366 loops=3)
                           Index Cond: (metadata_id = ANY ('{137,130,228,208,10,134,251,248,54,196,203,5,47,117,133,174,229,182,28,62,16,210,227,241,79,36,39,181,51,9,167,142,17,222,247,109,42,77,206,144,70,146,38,25,198,233,150,149,113,30,207,22,60,225,50,250,172,56,235,148,33,183,12,143,151,26,118,231,2,13,230,115,140,23,78,252,139,53,213,136,234,132,219,108,145,170,200,31,141,64,20,211,122,1,37,8,19,215,116,138,49,221,128,66,199,110,63,249,52,41,35,224,156,59,57,165,253,4,11,68,202,204,58,46,3,205,7,107,239,218,216,147,65,186,232,43,111,45,223,6,194,201,48,135,15,195,212,34,155,27,21,67,14,29,44,55,238,32,153,152,184,214,24,220,197,209,69,217,18,112,40,114,237,240,80,161,226}'::integer[]))
                           Heap Fetches: 1231
   ->  Index Scan using ix_statistics_short_term_statistic_id_start_ts on statistics_short_term  (cost=0.56..2.58 rows=1 width=60) (actual time=0.010..0.010 rows=1 loops=177)
         Index Cond: ((metadata_id = statistics_short_term_1.metadata_id) AND (start_ts = (max(statistics_short_term_1.start_ts))))
 Planning Time: 1.640 ms
 JIT:
   Functions: 21
   Options: Inlining false, Optimization false, Expressions true, Deforming true
   Timing: Generation 8.572 ms, Inlining 0.000 ms, Optimization 2.799 ms, Emission 59.284 ms, Total 70.655 ms
 Execution Time: 5539.016 ms
```
@bdraco any thoughts on this?
@momuskk have you had any luck with the latest version? 2023.7.1 as of this comment. I am noticing very high ongoing postgres (15.3-1.pgdg120+1) dockerised CPU usage. 

![image](https://github.com/home-assistant/core/assets/129476670/461de2c7-b647-4a93-bccf-4b3092f86854)

If you discover a faster query to replace and existing, I can help you convert it to sqlalchemy and test it on MariaDB and sqlite.  I don't have any production uses of postgresql so I'm not able to optimize for it.


In my second post is a new query for `has_events_context_ids_to_migrate()` that runs much faster and should be easy to do in sqlalchemy.

In my third post is a new statistics query that runs very fast, but will be more challenging to implement in sqlalchemy.  I found [this post](https://stackoverflow.com/questions/66025715/how-to-write-loose-index-scan-query-in-sqlalchemy) that may get us part of the way there.
The challenge is we need the query to still be supported and fast on MySQL and SQLite. 
```
SELECT COUNT(*)
FROM events 
WHERE events.context_id_bin IS NULL
```

That should be fast on sqlite since it will use the covering index either way but it will have to examine a lot of rows if they haven't been converted yet.

Its probably much slower on MariaDB though when the rows have not been converted yet.

I think it would make more sense to make a table of completed conversions (like we have for `schema_changes` and mark it as completed so we only have to check the table of conversions once. Than we wouldn't have to worry about what is faster on each engine since the query would only be to check the `schema_conversions` table.

Commit message: Save latest short term stat id for each metadata_id on each run

We currently have to do a algorithmically complex query to find the
latest short term stat id every five minutes when stats are run. To
avoid this we now save the stat id for each metadata_id in a new
table latest_statistics_short_term so we can do a very fast lookup
to find the data between runs.

This solution was arrived it after considering many other options
for months in #94435 as other solutions required per DB queries
which made the code significantly more complex

Patch to review:
diff --git a/homeassistant/components/recorder/db_schema.py b/homeassistant/components/recorder/db_schema.py
index e25c6d6dd5fbc..b5fa773f4ea43 100644
--- a/homeassistant/components/recorder/db_schema.py
+++ b/homeassistant/components/recorder/db_schema.py
@@ -84,6 +84,7 @@ class Base(DeclarativeBase):
 TABLE_STATISTICS_META = "statistics_meta"
 TABLE_STATISTICS_RUNS = "statistics_runs"
 TABLE_STATISTICS_SHORT_TERM = "statistics_short_term"
+TABLE_LATEST_STATISTICS_SHORT_TERM = "latest_statistics_short_term"
 
 STATISTICS_TABLES = ("statistics", "statistics_short_term")
 
@@ -105,6 +106,7 @@ class Base(DeclarativeBase):
     TABLE_STATISTICS_META,
     TABLE_STATISTICS_RUNS,
     TABLE_STATISTICS_SHORT_TERM,
+    TABLE_LATEST_STATISTICS_SHORT_TERM,
 ]
 
 TABLES_TO_CHECK = [
@@ -803,6 +805,29 @@ def __repr__(self) -> str:
         )
 
 
+class LatestStatisticsShortTerm(Base):
+    """Track latest short term statistics.
+
+    This table keeps track of the id of the latest short term statistic
+    for each metadata_id to avoid having a complex query to figure out where
+    we left off every five minutes when we build statistics.
+    """
+
+    __tablename__ = TABLE_LATEST_STATISTICS_SHORT_TERM
+    metadata_id: Mapped[int | None] = mapped_column(
+        Integer,
+        primary_key=True,
+    )
+    id: Mapped[int] = mapped_column(Integer)
+
+    def __repr__(self) -> str:
+        """Return string representation of instance for debugging."""
+        return (
+            f"<recorder.LatestStatisticsShortTerm(metadata_id={self.metadata_id},"
+            f" id='{self.id}')>"
+        )
+
+
 EVENT_DATA_JSON = type_coerce(
     EventData.shared_data.cast(JSONB_VARIANT_CAST), JSONLiteral(none_as_null=True)
 )
diff --git a/homeassistant/components/recorder/statistics.py b/homeassistant/components/recorder/statistics.py
index 005859b865b5b..a349c2e87f767 100644
--- a/homeassistant/components/recorder/statistics.py
+++ b/homeassistant/components/recorder/statistics.py
@@ -14,7 +14,7 @@
 from statistics import mean
 from typing import TYPE_CHECKING, Any, Literal, TypedDict, cast
 
-from sqlalchemy import Select, and_, bindparam, func, lambda_stmt, select, text
+from sqlalchemy import Select, and_, bindparam, delete, func, lambda_stmt, select, text
 from sqlalchemy.engine.row import Row
 from sqlalchemy.exc import SQLAlchemyError, StatementError
 from sqlalchemy.orm.session import Session
@@ -54,6 +54,7 @@
 )
 from .db_schema import (
     STATISTICS_TABLES,
+    LatestStatisticsShortTerm,
     Statistics,
     StatisticsBase,
     StatisticsRuns,
@@ -508,19 +509,23 @@ def _compile_statistics(
         platform_stats.extend(compiled.platform_stats)
         current_metadata.update(compiled.current_metadata)
 
+    new_short_term_stats: list[StatisticsBase] = []
+    updated_metadata_ids: set[int] = set()
     # Insert collected statistics in the database
     for stats in platform_stats:
         modified_statistic_id, metadata_id = statistics_meta_manager.update_or_add(
             session, stats["meta"], current_metadata
         )
-        if modified_statistic_id is not None:
+        if modified_statistic_id:
             modified_statistic_ids.add(modified_statistic_id)
-        _insert_statistics(
+        updated_metadata_ids.add(metadata_id)
+        if new_stat := _insert_statistics(
             session,
             StatisticsShortTerm,
             metadata_id,
             stats["stat"],
-        )
+        ):
+            new_short_term_stats.append(new_stat)
 
     if start.minute == 55:
         # A full hour is ready, summarize it
@@ -533,6 +538,20 @@ def _compile_statistics(
         if start.minute == 55:
             instance.hass.bus.fire(EVENT_RECORDER_HOURLY_STATISTICS_GENERATED)
 
+    if updated_metadata_ids:
+        # These are always the newest statistics, so we can update the latest
+        # statistics table without having to check the start_ts.
+        session.flush()  # populate the ids of the new StatisticsShortTerm rows
+        session.execute(
+            _delete_latest_statistics_short_term_by_metadata_ids_stmt(
+                updated_metadata_ids
+            ),
+        )
+        session.add_all(
+            LatestStatisticsShortTerm(metadata_id=new_stat.metadata_id, id=new_stat.id)
+            for new_stat in new_short_term_stats
+        )
+
     return modified_statistic_ids
 
 
@@ -566,16 +585,19 @@ def _insert_statistics(
     table: type[StatisticsBase],
     metadata_id: int,
     statistic: StatisticData,
-) -> None:
+) -> StatisticsBase | None:
     """Insert statistics in the database."""
     try:
-        session.add(table.from_stats(metadata_id, statistic))
+        stat = table.from_stats(metadata_id, statistic)
+        session.add(stat)
+        return stat
     except SQLAlchemyError:
         _LOGGER.exception(
             "Unexpected exception when inserting statistics %s:%s ",
             metadata_id,
             statistic,
         )
+        return None
 
 
 def _update_statistics(
@@ -1809,24 +1831,27 @@ def get_last_short_term_statistics(
     )
 
 
-def _latest_short_term_statistics_stmt(
-    metadata_ids: list[int],
+def _latest_short_term_statistics_from_latest_stmt(
+    metadata_ids: set[int],
+) -> StatementLambdaElement:
+    """Create the statement for finding the latest short term stat rows using LatestStatisticsShortTerm."""
+    return lambda_stmt(
+        lambda: select(*QUERY_STATISTICS_SHORT_TERM)
+        .select_from(LatestStatisticsShortTerm)
+        .join(
+            StatisticsShortTerm, LatestStatisticsShortTerm.id == StatisticsShortTerm.id
+        )
+        .filter(LatestStatisticsShortTerm.metadata_id.in_(metadata_ids))
+    )
+
+
+def _short_term_statistics_by_ids_stmt(
+    ids: set[int],
 ) -> StatementLambdaElement:
     """Create the statement for finding the latest short term stat rows."""
     return lambda_stmt(
-        lambda: select(*QUERY_STATISTICS_SHORT_TERM).join(
-            (
-                most_recent_statistic_row := (
-                    select(
-                        StatisticsShortTerm.metadata_id,
-                        func.max(StatisticsShortTerm.start_ts).label("start_max"),
-                    )
-                    .where(StatisticsShortTerm.metadata_id.in_(metadata_ids))
-                    .group_by(StatisticsShortTerm.metadata_id)
-                ).subquery()
-            ),
-            (StatisticsShortTerm.metadata_id == most_recent_statistic_row.c.metadata_id)
-            & (StatisticsShortTerm.start_ts == most_recent_statistic_row.c.start_max),
+        lambda: select(*QUERY_STATISTICS_SHORT_TERM).where(
+            StatisticsShortTerm.id.in_(ids)
         )
     )
 
@@ -1846,11 +1871,40 @@ def get_latest_short_term_statistics(
             )
         if not metadata:
             return {}
-        metadata_ids = _extract_metadata_and_discard_impossible_columns(metadata, types)
-        stmt = _latest_short_term_statistics_stmt(metadata_ids)
+        metadata_ids = set(
+            _extract_metadata_and_discard_impossible_columns(metadata, types)
+        )
+        stmt = _latest_short_term_statistics_from_latest_stmt(metadata_ids)
         stats = cast(
             Sequence[Row], execute_stmt_lambda_element(session, stmt, orm_rows=False)
         )
+        found_metadata_ids = {stat.metadata_id for stat in stats}
+        # If we do not have a row for a metadata_id, we need to find it in the table
+        # manually with a slower query. This should only happen once per metadata_id
+        # ever since the latest statistics table was introduced.
+        if missing_metadata_ids := metadata_ids - found_metadata_ids:
+            missing_ids: set[int] = set()
+            for metadata_id in missing_metadata_ids:
+                if (
+                    latest := cast(
+                        Sequence[Row],
+                        execute_stmt_lambda_element(
+                            session,
+                            _find_latest_short_term_statistic_stmt(metadata_id),
+                            orm_rows=False,
+                        ),
+                    )
+                ) and (missing_id := latest[0].id):
+                    missing_ids.add(missing_id)
+            if missing_ids:
+                stmt = _short_term_statistics_by_ids_stmt(missing_ids)
+                if stats_for_missing_rows := cast(
+                    Sequence[Row],
+                    execute_stmt_lambda_element(session, stmt, orm_rows=False),
+                ):
+                    stats = list(stats)
+                    stats.extend(stats_for_missing_rows)
+
         if not stats:
             return {}
 
@@ -2221,9 +2275,46 @@ def _import_statistics_with_session(
         else:
             _insert_statistics(session, table, metadata_id, stat)
 
+    if table != StatisticsShortTerm:
+        return True
+
+    session.execute(
+        _delete_latest_statistics_short_term_by_metadata_ids_stmt({metadata_id}),
+    )
+    if latest := cast(
+        Sequence[Row],
+        execute_stmt_lambda_element(
+            session, _find_latest_short_term_statistic_stmt(metadata_id), orm_rows=False
+        ),
+    ):
+        session.add(LatestStatisticsShortTerm(metadata_id=metadata_id, id=latest[0].id))
+
     return True
 
 
+def _find_latest_short_term_statistic_stmt(
+    metadata_id: int,
+) -> StatementLambdaElement:
+    """Find the latest short term statistics for a list of metadata_ids."""
+    return lambda_stmt(
+        lambda: select(
+            StatisticsShortTerm.id,
+            func.max(StatisticsShortTerm.start_ts).label("start_max"),
+        ).where(StatisticsShortTerm.metadata_id == metadata_id)
+    )
+
+
+def _delete_latest_statistics_short_term_by_metadata_ids_stmt(
+    metadata_ids: set[int],
+) -> StatementLambdaElement:
+    """Delete LatestStatisticsShortTerm rows."""
+    return lambda_stmt(
+        lambda: delete(LatestStatisticsShortTerm)
+        .where(LatestStatisticsShortTerm.metadata_id.in_(metadata_ids))
+        .execution_options(synchronize_session=False)
+    )
+
+
 @retryable_database_job("statistics")
 def import_statistics(
     instance: Recorder,


Code review comments:
diff_hunk: @@ -508,19 +509,23 @@ def _compile_statistics(
         platform_stats.extend(compiled.platform_stats)
         current_metadata.update(compiled.current_metadata)
 
+    new_short_term_stats: list[StatisticsBase] = []
+    updated_metadata_ids: set[int] = set()
     # Insert collected statistics in the database
     for stats in platform_stats:
         modified_statistic_id, metadata_id = statistics_meta_manager.update_or_add(
             session, stats["meta"], current_metadata
         )
-        if modified_statistic_id is not None:
+        if modified_statistic_id:
path: homeassistant/components/recorder/statistics.py
text: @author:
```suggestion
        if modified_statistic_id is not None:
```