fix: Elasticsearch Feature Registration (#296)

zabarn · Zach Barnett · web-flow · commit 83c7ca8ef9af · 2025-09-09T13:36:38.000-05:00
* [EAPC-19839] fix: Elasticsearch Feature Registration

* feat: update es test file with lazy_table_creation

---------

Co-authored-by: Zach Barnett &lt;zbarnett@expediagroup.com&gt;
diff --git a/sdk/python/feast/expediagroup/vectordb/elasticsearch_online_store.py b/sdk/python/feast/expediagroup/vectordb/elasticsearch_online_store.py
@@ -71,6 +71,12 @@ class ElasticsearchOnlineStoreConfig(FeastConfigBaseModel):
     password: str
     """ password to connect to Elasticsearch """
 
+    lazy_table_creation: Optional[bool] = False
+    """
+    If True, tables will be created during materialization, rather than registration.
+    Table deletion is not currently supported in this mode.
+    """
+
 
 class ElasticsearchConnectionManager:
     def __init__(self, online_config: RepoConfig):
diff --git a/sdk/python/feast/infra/online_stores/contrib/cassandra_online_store/cassandra_online_store.py b/sdk/python/feast/infra/online_stores/contrib/cassandra_online_store/cassandra_online_store.py
@@ -174,7 +174,7 @@ class CassandraOnlineStoreConfig(FeastConfigBaseModel):
 
     lazy_table_creation: Optional[bool] = False
     """
-    If True, tables will be created on during materialization, rather than registration.
+    If True, tables will be created during materialization, rather than registration.
     Table deletion is not currently supported in this mode.
     """
 
diff --git a/sdk/python/tests/expediagroup/test_elasticsearch_online_store.py b/sdk/python/tests/expediagroup/test_elasticsearch_online_store.py
@@ -48,8 +48,9 @@
 ]
 
 
-@pytest.fixture(scope="session")
-def repo_config(embedded_elasticsearch):
+@pytest.fixture(scope="session", params=[False, True], ids=["normal", "lazy"])
+def repo_config(request, embedded_elasticsearch):
+    """Parametrized repo config with lazy_table_creation=False and True"""
     return RepoConfig(
         registry=REGISTRY,
         project=PROJECT,
@@ -58,6 +59,7 @@ def repo_config(embedded_elasticsearch):
             endpoint=f"http://{embedded_elasticsearch['host']}:{embedded_elasticsearch['port']}",
             username=embedded_elasticsearch["username"],
             password=embedded_elasticsearch["password"],
+            lazy_table_creation=request.param,  # This will be False, then True
         ),
         offline_store=DaskOfflineStoreConfig(),
         entity_key_serialization_version=2,
@@ -75,24 +77,26 @@ def embedded_elasticsearch():
 
 
 class TestElasticsearchOnlineStore:
-    index_to_write = "index_write"
-    index_to_delete = "index_delete"
-    index_to_read = "index_read"
-    unavailable_index = "abc"
-
     @pytest.fixture(autouse=True)
     def setup_method(self, repo_config):
+        # Generate unique index names based on lazy_table_creation setting
+        lazy_suffix = "_lazy" if repo_config.online_store.lazy_table_creation else ""
+        self.index_to_write = f"index_write{lazy_suffix}"
+        self.index_to_delete = f"index_delete{lazy_suffix}"
+        self.index_to_read = f"index_read{lazy_suffix}"
+        self.unavailable_index = f"abc{lazy_suffix}"
+
         # Ensuring that the indexes created are dropped before the tests are run
         with ElasticsearchConnectionManager(repo_config.online_store) as es:
             # Dropping indexes if they exist
-            if es.indices.exists(index=self.index_to_delete):
-                es.indices.delete(index=self.index_to_delete)
-            if es.indices.exists(index=self.index_to_write):
-                es.indices.delete(index=self.index_to_write)
-            if es.indices.exists(index=self.index_to_read):
-                es.indices.delete(index=self.index_to_read)
-            if es.indices.exists(index=self.unavailable_index):
-                es.indices.delete(index=self.unavailable_index)
+            for index in [
+                self.index_to_delete,
+                self.index_to_write,
+                self.index_to_read,
+                self.unavailable_index,
+            ]:
+                if es.indices.exists(index=index):
+                    es.indices.delete(index=index)
 
         yield
 
