feat: Support cache_mode for registries

ntkathole · ntkathole · commit 021e9ea759bf · 2025-10-16T21:42:19.000+05:30
Signed-off-by: ntkathole &lt;nikhilkathole2683@gmail.com&gt;
diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py
@@ -1073,6 +1073,17 @@ def apply(
 
         self._registry.commit()
 
+        # Refresh the registry cache to ensure that changes are immediately visible
+        # This is especially important for UI and other clients that may be reading
+        # from the registry, as it ensures they see the updated state without waiting
+        # for the cache TTL to expire.
+        #
+        # Behavior by cache_mode:
+        # - sync mode: Immediate consistency - refresh after apply
+        # - thread mode: Eventual consistency - skip refresh, background thread handles it
+        if self.config.registry.cache_mode == "sync":
+            self.refresh_registry()
+
     def teardown(self):
         """Tears down all local and cloud resources for the feature store."""
         tables: List[FeatureView] = []
diff --git a/sdk/python/feast/infra/registry/sql.py b/sdk/python/feast/infra/registry/sql.py
@@ -234,9 +234,6 @@ class SqlRegistryConfig(RegistryConfig):
     sqlalchemy_config_kwargs: Dict[str, Any] = {"echo": False}
     """ Dict[str, Any]: Extra arguments to pass to SQLAlchemy.create_engine. """
 
-    cache_mode: StrictStr = "sync"
-    """ str: Cache mode type, Possible options are sync and thread(asynchronous caching using threading library)"""
-
     thread_pool_executor_worker_count: StrictInt = 0
     """ int: Number of worker threads to use for asynchronous caching in SQL Registry. If set to 0, it doesn't use ThreadPoolExecutor. """
 
diff --git a/sdk/python/feast/repo_config.py b/sdk/python/feast/repo_config.py
@@ -152,6 +152,12 @@ class RegistryConfig(FeastBaseModel):
      set to infinity by setting TTL to 0 seconds, which means the cache will only be loaded once and will never
      expire. Users can manually refresh the cache by calling feature_store.refresh_registry() """
 
+    cache_mode: StrictStr = "sync"
+    """str: Cache mode type. Possible options are 'sync' (immediate refresh after each write operation) and
+     'thread' (asynchronous background refresh at cache_ttl_seconds intervals). In 'sync' mode, registry changes
+     are immediately visible. In 'thread' mode, changes may take up to
+     cache_ttl_seconds to be visible."""
+
     s3_additional_kwargs: Optional[Dict[str, str]] = None
     """ Dict[str, str]: Extra arguments to pass to boto3 when writing the registry file to S3. """
 
diff --git a/sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py b/sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py
@@ -1,5 +1,6 @@
 from datetime import datetime, timedelta
 from tempfile import mkstemp
+from unittest.mock import patch
 
 import pytest
 from pytest_lazyfixture import lazy_fixture
@@ -18,7 +19,7 @@
 from feast.permissions.action import AuthzedAction
 from feast.permissions.permission import Permission
 from feast.permissions.policy import RoleBasedPolicy
-from feast.repo_config import RepoConfig
+from feast.repo_config import RegistryConfig, RepoConfig
 from feast.stream_feature_view import stream_feature_view
 from feast.types import Array, Bytes, Float32, Int64, String, ValueType, from_value_type
 from tests.integration.feature_repos.universal.feature_views import TAGS
@@ -797,3 +798,67 @@ def feature_store_with_local_registry():
             entity_key_serialization_version=3,
         )
     )
+
+
+@pytest.mark.parametrize(
+    "test_feature_store",
+    [lazy_fixture("feature_store_with_local_registry")],
+)
+def test_apply_refreshes_registry_cache_sync_mode(test_feature_store):
+    """Test that apply() refreshes registry cache when cache_mode is 'sync' (default)"""
+    # Create a simple entity (no FeatureView to avoid file path issues)
+    entity = Entity(name="test_entity", join_keys=["id"])
+
+    # Mock the refresh_registry method to verify it's called
+    with patch.object(test_feature_store, "refresh_registry") as mock_refresh:
+        # Apply the entity
+        test_feature_store.apply([entity])
+
+        # Verify refresh_registry was called once (due to sync mode)
+        mock_refresh.assert_called_once()
+
+    test_feature_store.teardown()
+
+
+@pytest.mark.parametrize(
+    "test_feature_store",
+    [lazy_fixture("feature_store_with_local_registry")],
+)
+def test_apply_skips_refresh_registry_cache_thread_mode(test_feature_store):
+    """Test that apply() skips registry refresh when cache_mode is 'thread'"""
+    # Create a simple entity
+    entity = Entity(name="test_entity", join_keys=["id"])
+
+    # Temporarily change cache_mode to 'thread'
+    original_cache_mode = test_feature_store.config.registry.cache_mode
+    test_feature_store.config.registry.cache_mode = "thread"
+
+    try:
+        # Mock the refresh_registry method to verify it's NOT called
+        with patch.object(test_feature_store, "refresh_registry") as mock_refresh:
+            # Apply the entity
+            test_feature_store.apply([entity])
+
+            # Verify refresh_registry was NOT called (due to thread mode)
+            mock_refresh.assert_not_called()
+    finally:
+        # Restore original cache_mode
+        test_feature_store.config.registry.cache_mode = original_cache_mode
+
+    test_feature_store.teardown()
+
+
+def test_registry_config_cache_mode_default():
+    """Test that RegistryConfig has cache_mode with default value 'sync'"""
+    config = RegistryConfig()
+    assert hasattr(config, "cache_mode")
+    assert config.cache_mode == "sync"
+
+
+def test_registry_config_cache_mode_can_be_set():
+    """Test that RegistryConfig cache_mode can be set to different values"""
+    config = RegistryConfig(cache_mode="thread")
+    assert config.cache_mode == "thread"
+
+    config = RegistryConfig(cache_mode="sync")
+    assert config.cache_mode == "sync"
