@@ -919,6 +919,16 @@ def refresh(self, project: Optional[str] = None):
         """Refreshes the state of the registry cache by fetching the registry state from the remote registry store."""
         raise NotImplementedError
 
+    def is_cache_valid(self) -> bool:
+        """Check whether the registry's local cache is still within its TTL.
+
+        Returns True if cached data can be used without a refresh.
+        Subclasses that support caching should override this.
+        Registries without caching always return False (every read goes
+        to the backing store).
+        """
+        return False
+
     # Lineage operations
     def get_registry_lineage(
         self,
 
@@ -442,32 +442,26 @@ def refresh(self, project: Optional[str] = None):
         except Exception as e:
             logger.debug(f"Error while refreshing registry: {e}", exc_info=True)
 
+    def is_cache_valid(self) -> bool:
+        if (
+            self.cached_registry_proto is None
+            or self.cached_registry_proto == RegistryProto()
+        ):
+            return False
+        if (
+            not hasattr(self, "cached_registry_proto_created")
+            or self.cached_registry_proto_created is None
+        ):
+            return False
+        if self.cached_registry_proto_ttl.total_seconds() > 0 and _utc_now() > (
+            self.cached_registry_proto_created + self.cached_registry_proto_ttl
+        ):
+            return False
+        return True
+
     def _refresh_cached_registry_if_necessary(self):
         if self.cache_mode == "sync":
-
-            def is_cache_expired():
-                if (
-                    self.cached_registry_proto is None
-                    or self.cached_registry_proto == RegistryProto()
-                ):
-                    return True
-
-                # Cache is expired if creation time is None
-                if (
-                    not hasattr(self, "cached_registry_proto_created")
-                    or self.cached_registry_proto_created is None
-                ):
-                    return True
-
-                # Cache is expired if TTL > 0 and current time exceeds creation + TTL
-                if self.cached_registry_proto_ttl.total_seconds() > 0 and _utc_now() > (
-                    self.cached_registry_proto_created + self.cached_registry_proto_ttl
-                ):
-                    return True
-
-                return False
-
-            if is_cache_expired():
+            if not self.is_cache_valid():
                 if not self._refresh_lock.acquire(blocking=False):
                     logger.debug(
                         "Skipping refresh if lock is already held by another thread"
 
@@ -1287,6 +1287,17 @@ def refresh(self, project: Optional[str] = None):
         """Refreshes the state of the registry cache by fetching the registry state from the remote registry store."""
         self._get_registry_proto(project=project, allow_cache=False)
 
+    def is_cache_valid(self) -> bool:
+        if self.cached_registry_proto_created is None:
+            return False
+        if (
+            self.cached_registry_proto_ttl.total_seconds() > 0
+            and _utc_now()
+            > self.cached_registry_proto_created + self.cached_registry_proto_ttl
+        ):
+            return False
+        return True
+
     def teardown(self):
         """Tears down (removes) the registry."""
         self._registry_store.teardown()
 
@@ -1,6 +1,8 @@
 import copy
 import itertools
+import logging
 import os
+import threading
 import typing
 import warnings
 from collections import Counter, defaultdict
@@ -1380,6 +1382,96 @@ def _get_online_request_context(
     )
 
 
+_feature_resolution_cache: Dict[tuple, tuple] = {}
+_feature_resolution_cache_lock = threading.Lock()
+_feature_resolution_registry_ts: Optional[datetime] = None
+
+_logger = logging.getLogger(__name__)
+
+
+def _get_cached_request_context(
+    registry,
+    project: str,
+    features: Union[List[str], "FeatureService"],
+    full_feature_names: bool,
+):
+    """Return the output of _get_online_request_context plus resolved ODFV
+    entity join keys, using a cache that is invalidated whenever the
+    registry refreshes or its cache TTL expires."""
+    from feast.feature_service import FeatureService as _FS
+
+    global _feature_resolution_cache, _feature_resolution_registry_ts
+
+    registry_ts = getattr(registry, "cached_registry_proto_created", None)
+
+    if isinstance(features, _FS):
+        features_key: tuple = ("__fs__", features.name)
+    else:
+        features_key = tuple(features)
+
+    cache_key = (features_key, project, full_feature_names)
+
+    is_cache_valid = getattr(registry, "is_cache_valid", None)
+    registry_cache_valid = is_cache_valid() if callable(is_cache_valid) else False
+
+    if registry_ts is not None and registry_cache_valid:
+        with _feature_resolution_cache_lock:
+            if registry_ts != _feature_resolution_registry_ts:
+                _feature_resolution_cache.clear()
+                _feature_resolution_registry_ts = registry_ts
+                _logger.debug("Feature resolution cache cleared (registry refreshed)")
+            else:
+                cached = _feature_resolution_cache.get(cache_key)
+                if cached is not None:
+                    return cached
+
+    ctx = _get_online_request_context(registry, project, features, full_feature_names)
+
+    (
+        feature_refs,
+        requested_on_demand_feature_views,
+        entity_name_to_join_key_map,
+        entity_type_map,
+        join_keys_set,
+        grouped_refs,
+        requested_result_row_names,
+        needed_request_data,
+        entityless_case,
+    ) = ctx
+
+    odfv_join_keys: set = set()
+    for on_demand_feature_view in requested_on_demand_feature_views:
+        entities_for_odfv = getattr(on_demand_feature_view, "entities", [])
+        if len(entities_for_odfv) > 0 and isinstance(entities_for_odfv[0], str):
+            entities_for_odfv = [
+                registry.get_entity(entity_name, project, allow_cache=True)
+                for entity_name in entities_for_odfv
+            ]
+        odfv_join_keys.update(entities_for_odfv)
+
+    result = (
+        feature_refs,
+        requested_on_demand_feature_views,
+        entity_name_to_join_key_map,
+        entity_type_map,
+        join_keys_set | odfv_join_keys,
+        grouped_refs,
+        frozenset(requested_result_row_names),
+        needed_request_data,
+        entityless_case,
+    )
+
+    registry_ts_after = getattr(registry, "cached_registry_proto_created", None)
+    if registry_ts_after is not None:
+        with _feature_resolution_cache_lock:
+            if registry_ts_after != _feature_resolution_registry_ts:
+                _feature_resolution_cache.clear()
+                _feature_resolution_registry_ts = registry_ts_after
+            _feature_resolution_cache[cache_key] = result
+
+    return result
+
+
 def _prepare_entities_to_read_from_online_store(
     registry,
     project,
@@ -1399,10 +1491,13 @@ def _prepare_entities_to_read_from_online_store(
         entity_type_map,
         join_keys_set,
         grouped_refs,
-        requested_result_row_names,
+        requested_result_row_names_frozen,
         needed_request_data,
         entityless_case,
-    ) = _get_online_request_context(registry, project, features, full_feature_names)
+    ) = _get_cached_request_context(registry, project, features, full_feature_names)
+
+    # Mutable copy — downstream code adds join keys to this set.
+    requested_result_row_names = set(requested_result_row_names_frozen)
 
     # Extract Sequence from RepeatedValue Protobuf.
     entity_value_lists: Dict[str, Union[List[Any], List[ValueProto]]] = {
@@ -1424,23 +1519,6 @@ def _prepare_entities_to_read_from_online_store(
 
     num_rows = _validate_entity_values(entity_proto_values)
 
-    odfv_entities: List[Entity] = []
-    request_source_keys: List[str] = []
-    for on_demand_feature_view in requested_on_demand_feature_views:
-        entities_for_odfv = getattr(on_demand_feature_view, "entities", [])
-        if len(entities_for_odfv) > 0 and isinstance(entities_for_odfv[0], str):
-            entities_for_odfv = [
-                registry.get_entity(entity_name, project, allow_cache=True)
-                for entity_name in entities_for_odfv
-            ]
-        odfv_entities.extend(entities_for_odfv)
-        for source in on_demand_feature_view.source_request_sources:
-            source_schema = on_demand_feature_view.source_request_sources[source].schema
-            for column in source_schema:
-                request_source_keys.append(column.name)
-
-    join_keys_set.update(set(odfv_entities))
-
     join_key_values: Dict[str, List[ValueProto]] = {}
     request_data_features: Dict[str, List[ValueProto]] = {}
     # Entity rows may be either entities or request data.