Avoid joining states_meta for statistics queries (#89941)

homeassistant/components/recorder/history/modern.py

@@ -142,8 +142,8 @@ def _ignore_domains_filter(query: Query) -> Query:
 def _significant_states_stmt(
     start_time: datetime,
     end_time: datetime | None,
-    entity_ids: list[str] | None,
     metadata_ids: list[int] | None,
+    metadata_ids_in_significant_domains: list[int],
     filters: Filters | None,
     significant_changes_only: bool,
     no_attributes: bool,
@@ -153,17 +153,23 @@ def _significant_states_stmt(
         no_attributes, include_last_changed=not significant_changes_only
     )
     join_states_meta = False
-    if (
-        entity_ids
-        and len(entity_ids) == 1
-        and significant_changes_only
-        and split_entity_id(entity_ids[0])[0] not in SIGNIFICANT_DOMAINS
-    ):
+    if metadata_ids and significant_changes_only:
+        # Since we are filtering on entity_id (metadata_id) we can avoid
+        # the join of the states_meta table since we already know which
+        # metadata_ids are in the significant domains.
         stmt += lambda q: q.filter(
-            (States.last_changed_ts == States.last_updated_ts)
+            States.metadata_id.in_(metadata_ids_in_significant_domains)
+            | (States.last_changed_ts == States.last_updated_ts)
             | States.last_changed_ts.is_(None)
         )
     elif significant_changes_only:
+        # This is the case where we are not filtering on entity_id
+        # so we need to join the states_meta table to filter out
+        # the domains we do not care about. This query path was
+        # only used by the old history page to show all entities
+        # in the UI. The new history page filters on entity_id
+        # so this query path is not used anymore except for third
+        # party integrations that use the history API.
         stmt += lambda q: q.filter(
             or_(
                 *[
@@ -235,6 +241,7 @@ def get_significant_states_with_session(
     """
     metadata_ids: list[int] | None = None
     entity_id_to_metadata_id: dict[str, int | None] | None = None
+    metadata_ids_in_significant_domains: list[int] = []
     if entity_ids:
         instance = recorder.get_instance(hass)
         if not (
@@ -243,11 +250,18 @@ def get_significant_states_with_session(
             )
         ) or not (metadata_ids := extract_metadata_ids(entity_id_to_metadata_id)):
             return {}
+        if significant_changes_only:
+            metadata_ids_in_significant_domains = [
+                metadata_id
+                for entity_id, metadata_id in entity_id_to_metadata_id.items()
+                if metadata_id is not None
+                and split_entity_id(entity_id)[0] in SIGNIFICANT_DOMAINS
+            ]
     stmt = _significant_states_stmt(
         start_time,
         end_time,
-        entity_ids,
         metadata_ids,
+        metadata_ids_in_significant_domains,
         filters,
         significant_changes_only,
         no_attributes,