From da129bf5e2db41a88a73bbb6f41adc95d88e9253 Mon Sep 17 00:00:00 2001 From: Pavel Alekseev Date: Sat, 22 Aug 2026 17:02:04 +0300 Subject: [PATCH 1/2] feat: Add feature view versioning support to MongoDB online store MongoDB raised VersionedOnlineReadNotSupported for any version-qualified reference, so feature view versioning could not be used on MongoDB at all. Unlike the stores that already support this, MongoDB keeps one collection per project and stores each feature view inside it as a sub-document key, under features. and event_timestamps.. Versioning the collection name would split one entity's document across collections and duplicate every vector index, so the version qualifies the document namespace instead, following redis.py rather than sqlite.py. Adds _versioned_fv_name() and applies it on the write path, both read paths, the document converter, the vector search path, update(), and the Atlas vector index names and paths, so each version gets its own index. teardown() is unchanged: it drops the whole collection. Also registers MongoDBOnlineStore in the allow-list consulted by OnlineStore._is_versioned_read_supported(), without which a versioned read still raises regardless of what the store does. With versioning disabled the document keys are byte-for-byte what they were, and the two new parameters both default to the previous behaviour. Signed-off-by: Pavel Alekseev --- sdk/python/feast/errors.py | 2 +- .../mongodb_online_store/mongodb.py | 79 ++-- .../feast/infra/online_stores/online_store.py | 4 + .../online_store/test_mongodb_versioning.py | 357 ++++++++++++++++++ 4 files changed, 415 insertions(+), 27 deletions(-) create mode 100644 sdk/python/tests/unit/infra/online_store/test_mongodb_versioning.py diff --git a/sdk/python/feast/errors.py b/sdk/python/feast/errors.py index 515a6c39b11..0f7fac4e6de 100644 --- a/sdk/python/feast/errors.py +++ b/sdk/python/feast/errors.py @@ -142,7 +142,7 @@ class VersionedOnlineReadNotSupported(FeastError): def __init__(self, store_name: str, version: int): super().__init__( f"Versioned feature reads (@v{version}) are not yet supported by {store_name}. " - f"Currently only SQLite, PostgreSQL, MySQL, FAISS, Redis, and DynamoDB support version-qualified feature references. " + f"Currently only SQLite, PostgreSQL, MySQL, FAISS, Redis, DynamoDB, and MongoDB support version-qualified feature references. " ) diff --git a/sdk/python/feast/infra/online_stores/mongodb_online_store/mongodb.py b/sdk/python/feast/infra/online_stores/mongodb_online_store/mongodb.py index 16dd572bf5b..546e7f8e7ce 100644 --- a/sdk/python/feast/infra/online_stores/mongodb_online_store/mongodb.py +++ b/sdk/python/feast/infra/online_stores/mongodb_online_store/mongodb.py @@ -27,6 +27,7 @@ FilterType, ) from feast.infra.key_encoding_utils import deserialize_entity_key, serialize_entity_key +from feast.infra.online_stores.helpers import compute_versioned_name from feast.infra.online_stores.online_store import OnlineStore from feast.infra.online_stores.vector_store import VectorStoreConfig from feast.infra.supported_async_methods import SupportedAsyncMethods @@ -43,6 +44,14 @@ DRIVER_METADATA = DriverInfo(name="Feast", version=feast.version.get_version()) + +def _versioned_fv_name(table: FeatureView, config: RepoConfig) -> str: + """Return the feature view name with a version suffix when versioning is enabled.""" + return compute_versioned_name( + table, config.registry.enable_online_feature_view_versioning + ) + + _MONGO_COMPARISON_OPS: Dict[str, str] = { "eq": "$eq", "ne": "$ne", @@ -186,6 +195,7 @@ def _build_write_ops( ``collection.bulk_write(ops, ordered=False)`` (sync) or ``await collection.bulk_write(ops, ordered=False)`` (async). """ + fv_name = _versioned_fv_name(table, config) ops = [] for entity_key, proto_values, event_timestamp, created_timestamp in data: entity_id = serialize_entity_key( @@ -193,13 +203,13 @@ def _build_write_ops( entity_key_serialization_version=config.entity_key_serialization_version, ) feature_updates = { - f"features.{table.name}.{field}": feast_value_type_to_python_type(val) + f"features.{fv_name}.{field}": feast_value_type_to_python_type(val) for field, val in proto_values.items() } update = { "$set": { **feature_updates, - f"event_timestamps.{table.name}": event_timestamp, + f"event_timestamps.{fv_name}": event_timestamp, "created_timestamp": created_timestamp, }, } @@ -261,6 +271,7 @@ def online_read( List of tuples (event_timestamp, feature_dict) for each entity key """ clxn = self._get_collection(config) + fv_name = _versioned_fv_name(table, config) ids = [ serialize_entity_key( @@ -273,19 +284,19 @@ def online_read( query_filter = {"_id": {"$in": ids}} projection = { "_id": 1, - f"event_timestamps.{table.name}": 1, + f"event_timestamps.{fv_name}": 1, } if requested_features: projection.update( - {f"features.{table.name}.{x}": 1 for x in requested_features} + {f"features.{fv_name}.{x}": 1 for x in requested_features} ) else: - projection[f"features.{table.name}"] = 1 + projection[f"features.{fv_name}"] = 1 cursor = clxn.find(query_filter, projection=projection) docs = {doc["_id"]: doc for doc in cursor} - return self._convert_raw_docs_to_proto(ids, docs, table) + return self._convert_raw_docs_to_proto(ids, docs, table, fv_name) def retrieve_online_documents_v2( self, @@ -327,6 +338,7 @@ def retrieve_online_documents_v2( ) clxn = self._get_collection(config) + fv_name = _versioned_fv_name(table, config) # Identify the vector field on this feature view vector_fields = [f for f in table.features if f.vector_index] @@ -335,8 +347,8 @@ def retrieve_online_documents_v2( f"Feature view '{table.name}' has no fields with vector_index=True." ) vector_field = vector_fields[0] - path = f"features.{table.name}.{vector_field.name}" - idx_name = self._vector_search_index_name(table.name, vector_field.name) + path = f"features.{fv_name}.{vector_field.name}" + idx_name = self._vector_search_index_name(fv_name, vector_field.name) # BSON cannot encode numpy float types — ensure native Python floats. query_vector = [float(v) for v in embedding] @@ -350,7 +362,7 @@ def retrieve_online_documents_v2( "limit": top_k, } - mql_filter = MongoDBFilterTranslator(table.name).translate(filters) + mql_filter = MongoDBFilterTranslator(fv_name).translate(filters) if mql_filter: vector_search_stage["filter"] = mql_filter @@ -384,10 +396,10 @@ def retrieve_online_documents_v2( ) # Event timestamp - event_ts = doc.get("event_timestamps", {}).get(table.name) + event_ts = doc.get("event_timestamps", {}).get(fv_name) # Build feature dict from raw doc values - fv_features = doc.get("features", {}).get(table.name, {}) + fv_features = doc.get("features", {}).get(fv_name, {}) # Convert raw values → ValueProto for each requested feature feature_dict: Dict[str, ValueProto] = {} @@ -450,22 +462,24 @@ def update( raise RuntimeError(f"{config.online_store.type = }. It must be mongodb.") online_config = config.online_store + versioning = config.registry.enable_online_feature_view_versioning clxn = self._get_collection(repo_config=config) # --- Remove deleted feature views (data + vector search indexes) --- if tables_to_delete: unset_fields = {} for fv in tables_to_delete: - unset_fields[f"features.{fv.name}"] = "" - unset_fields[f"event_timestamps.{fv.name}"] = "" + deleted_name = compute_versioned_name(fv, versioning) + unset_fields[f"features.{deleted_name}"] = "" + unset_fields[f"event_timestamps.{deleted_name}"] = "" clxn.update_many({}, {"$unset": unset_fields}) if online_config.vector_enabled: - self._drop_vector_indexes_for_tables(clxn, tables_to_delete) + self._drop_vector_indexes_for_tables(clxn, tables_to_delete, versioning) # --- Create vector search indexes for kept feature views --- if online_config.vector_enabled: - self._ensure_vector_indexes(clxn, tables_to_keep, online_config) + self._ensure_vector_indexes(clxn, tables_to_keep, online_config, versioning) # Note: entities_to_delete contains Entity definitions (metadata), not entity instances. # Like other online stores, we don't need to do anything with entities_to_delete here. @@ -554,12 +568,14 @@ def _drop_vector_indexes_for_tables( self, collection: Collection, tables: Sequence[FeatureView], + enable_versioning: bool = False, ) -> None: """Drop all Atlas vector search indexes belonging to the given feature views.""" existing = {idx["name"] for idx in collection.list_search_indexes()} for fv in tables: + versioned_name = compute_versioned_name(fv, enable_versioning) for field in fv.features: - idx_name = self._vector_search_index_name(fv.name, field.name) + idx_name = self._vector_search_index_name(versioned_name, field.name) if idx_name in existing: logger.info("Dropping vector search index: %s", idx_name) collection.drop_search_index(idx_name) @@ -569,6 +585,7 @@ def _ensure_vector_indexes( collection: Collection, tables: Sequence[Union[BatchFeatureView, StreamFeatureView, FeatureView]], online_config: MongoDBOnlineStoreConfig, + enable_versioning: bool = False, ) -> None: """Create Atlas vector search indexes for vector-indexed fields if they don't exist. @@ -586,14 +603,15 @@ def _ensure_vector_indexes( existing = {idx["name"] for idx in collection.list_search_indexes()} for fv in tables: + versioned_name = compute_versioned_name(fv, enable_versioning) vector_fields = [f for f in fv.features if f.vector_index] for field in vector_fields: - idx_name = self._vector_search_index_name(fv.name, field.name) + idx_name = self._vector_search_index_name(versioned_name, field.name) if idx_name in existing: logger.debug("Vector search index '%s' already exists", idx_name) continue - path = f"features.{fv.name}.{field.name}" + path = f"features.{versioned_name}.{field.name}" num_dimensions = field.vector_length if not num_dimensions: raise ValueError( @@ -691,7 +709,10 @@ def async_supported(self) -> SupportedAsyncMethods: @staticmethod def _convert_raw_docs_to_proto( - ids: list[bytes], docs: dict[bytes, Any], table: FeatureView + ids: list[bytes], + docs: dict[bytes, Any], + table: FeatureView, + fv_name: Optional[str] = None, ) -> List[Tuple[Optional[datetime], Optional[dict[str, ValueProto]]]]: """Optimized converting values in documents retrieved from MongoDB (BSON) into ValueProto types. @@ -707,9 +728,14 @@ def _convert_raw_docs_to_proto( ids: sorted list of the serialized entity ids requested. docs: results of collection find. table: The FeatureView of the read, providing the types. + fv_name: The document namespace the feature view was written + under, which carries a ``_v{N}`` suffix when feature view + versioning is enabled. Defaults to ``table.name``. Returns: List of tuples (event_timestamp, feature_dict) for each entity key """ + fv_name = fv_name or table.name + feature_type_map = { feature.name: feature.dtype.to_value_type() for feature in table.features } @@ -722,7 +748,7 @@ def _convert_raw_docs_to_proto( for entity_id in ids: doc = docs.get(entity_id) - feature_dict = doc.get("features", {}).get(table.name, {}) if doc else {} + feature_dict = doc.get("features", {}).get(fv_name, {}) if doc else {} # For each expected feature, append its value or None for feature_name in feature_type_map: @@ -750,12 +776,12 @@ def _convert_raw_docs_to_proto( # Entity document exists (written by some other feature view), but # this specific feature view was never written → treat as not found. - fv_features = doc.get("features", {}).get(table.name) + fv_features = doc.get("features", {}).get(fv_name) if fv_features is None: results.append((None, None)) continue - ts = doc.get("event_timestamps", {}).get(table.name) + ts = doc.get("event_timestamps", {}).get(fv_name) row_features = { feature_name: proto_feature_columns[feature_name][i] @@ -785,6 +811,7 @@ async def online_read_async( List of tuples (event_timestamp, feature_dict) for each entity key """ clxn = await self._get_collection_async(config) + fv_name = _versioned_fv_name(table, config) # Serialize entity keys ids = [ @@ -798,20 +825,20 @@ async def online_read_async( query_filter = {"_id": {"$in": ids}} projection = { "_id": 1, - f"event_timestamps.{table.name}": 1, + f"event_timestamps.{fv_name}": 1, } if requested_features: projection.update( - {f"features.{table.name}.{x}": 1 for x in requested_features} + {f"features.{fv_name}.{x}": 1 for x in requested_features} ) else: - projection[f"features.{table.name}"] = 1 + projection[f"features.{fv_name}"] = 1 cursor = clxn.find(query_filter, projection=projection) docs = {doc["_id"]: doc async for doc in cursor} # Convert to proto format - return self._convert_raw_docs_to_proto(ids, docs, table) + return self._convert_raw_docs_to_proto(ids, docs, table, fv_name) async def online_write_batch_async( self, diff --git a/sdk/python/feast/infra/online_stores/online_store.py b/sdk/python/feast/infra/online_stores/online_store.py index cdf06639fe0..57d31bfa233 100644 --- a/sdk/python/feast/infra/online_stores/online_store.py +++ b/sdk/python/feast/infra/online_stores/online_store.py @@ -315,6 +315,10 @@ def _is_versioned_read_supported(self) -> bool: "feast.infra.online_stores.milvus_online_store.milvus", "MilvusOnlineStore", ), + ( + "feast.infra.online_stores.mongodb_online_store.mongodb", + "MongoDBOnlineStore", + ), ): try: import importlib diff --git a/sdk/python/tests/unit/infra/online_store/test_mongodb_versioning.py b/sdk/python/tests/unit/infra/online_store/test_mongodb_versioning.py new file mode 100644 index 00000000000..a740577fd0a --- /dev/null +++ b/sdk/python/tests/unit/infra/online_store/test_mongodb_versioning.py @@ -0,0 +1,357 @@ +"""Unit tests for MongoDB online store feature view versioning.""" + +# ruff: noqa: E402 + +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock + +import pytest + +pytest.importorskip("pymongo") + +from feast import Entity, FeatureView, Field, FileSource, RepoConfig +from feast.infra.online_stores.mongodb_online_store.mongodb import ( + MongoDBOnlineStore, + _versioned_fv_name, +) +from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto +from feast.protos.feast.types.Value_pb2 import Value as ValueProto +from feast.types import Array, Float32, Int64 +from feast.value_type import ValueType + + +def _make_feature_view(name="driver_stats", version_number=None, version_tag=None): + entity = Entity( + name="driver_id", + join_keys=["driver_id"], + value_type=ValueType.INT64, + ) + fv = FeatureView( + name=name, + entities=[entity], + ttl=timedelta(days=1), + schema=[Field(name="trips_today", dtype=Int64)], + source=FileSource(path="fake.parquet", timestamp_field="event_timestamp"), + ) + if version_number is not None: + fv.current_version_number = version_number + if version_tag is not None: + fv.projection.version_tag = version_tag + return fv + + +def _make_config(project="test_project", versioning=False, vector_enabled=False): + """Build a real RepoConfig, so the registry flag is exercised rather than mocked.""" + return RepoConfig( + project=project, + provider="local", + online_store={ + "type": "mongodb", + "connection_string": "mongodb://localhost:27017", + "vector_enabled": vector_enabled, + }, + registry={ + "path": "memory://", + "enable_online_feature_view_versioning": versioning, + }, + entity_key_serialization_version=3, + ) + + +def _entity_key(driver_id=1): + key = EntityKeyProto() + key.join_keys.append("driver_id") + key.entity_values.append(ValueProto(int64_val=driver_id)) + return key + + +class TestVersionedFvName: + """_versioned_fv_name produces the document namespace the store writes under.""" + + def test_no_versioning(self): + fv = _make_feature_view() + assert _versioned_fv_name(fv, _make_config(versioning=False)) == "driver_stats" + + def test_versioning_disabled_ignores_version(self): + fv = _make_feature_view(version_number=3) + assert _versioned_fv_name(fv, _make_config(versioning=False)) == "driver_stats" + + def test_versioning_enabled_no_version_set(self): + fv = _make_feature_view() + assert _versioned_fv_name(fv, _make_config(versioning=True)) == "driver_stats" + + def test_versioning_enabled_with_current_version_number(self): + fv = _make_feature_view(version_number=2) + assert ( + _versioned_fv_name(fv, _make_config(versioning=True)) == "driver_stats_v2" + ) + + def test_version_zero_no_suffix(self): + fv = _make_feature_view(version_number=0) + assert _versioned_fv_name(fv, _make_config(versioning=True)) == "driver_stats" + + def test_projection_version_tag_takes_priority(self): + fv = _make_feature_view(version_number=1, version_tag=3) + assert ( + _versioned_fv_name(fv, _make_config(versioning=True)) == "driver_stats_v3" + ) + + def test_projection_version_tag_zero_no_suffix(self): + fv = _make_feature_view(version_tag=0, version_number=3) + assert _versioned_fv_name(fv, _make_config(versioning=True)) == "driver_stats" + + +class TestWritePathVersioning: + """Writes land under the versioned namespace inside the shared collection.""" + + @staticmethod + def _set_doc(fv, config): + ts = datetime(2024, 1, 1, tzinfo=timezone.utc) + data = [(_entity_key(), {"trips_today": ValueProto(int64_val=7)}, ts, ts)] + ops = MongoDBOnlineStore._build_write_ops(config, fv, data) + return ops[0]._doc["$set"] + + def test_unversioned_write_uses_plain_name(self): + doc = self._set_doc(_make_feature_view(), _make_config(versioning=False)) + assert "features.driver_stats.trips_today" in doc + assert "event_timestamps.driver_stats" in doc + + def test_versioned_write_uses_suffixed_name(self): + doc = self._set_doc( + _make_feature_view(version_number=2), _make_config(versioning=True) + ) + assert "features.driver_stats_v2.trips_today" in doc + assert "event_timestamps.driver_stats_v2" in doc + assert "features.driver_stats.trips_today" not in doc + + def test_two_versions_write_to_different_namespaces(self): + config = _make_config(versioning=True) + doc_v1 = self._set_doc(_make_feature_view(version_number=1), config) + doc_v2 = self._set_doc(_make_feature_view(version_number=2), config) + assert set(doc_v1) != set(doc_v2) + + def test_versioning_disabled_ignores_version_on_write(self): + doc = self._set_doc( + _make_feature_view(version_number=2), _make_config(versioning=False) + ) + assert "features.driver_stats.trips_today" in doc + + +class TestReadPathVersioning: + """Reads project the versioned namespace, so a versioned read cannot see v1 data.""" + + @staticmethod + def _projection(fv, config, requested=None): + store = MongoDBOnlineStore() + collection = MagicMock() + collection.find.return_value = [] + store._get_collection = MagicMock(return_value=collection) + store.online_read( + config=config, + table=fv, + entity_keys=[_entity_key()], + requested_features=requested, + ) + return collection.find.call_args.kwargs["projection"] + + def test_unversioned_read_projects_plain_name(self): + projection = self._projection(_make_feature_view(), _make_config()) + assert "features.driver_stats" in projection + assert "event_timestamps.driver_stats" in projection + + def test_versioned_read_projects_suffixed_name(self): + projection = self._projection( + _make_feature_view(version_number=2), _make_config(versioning=True) + ) + assert "features.driver_stats_v2" in projection + assert "event_timestamps.driver_stats_v2" in projection + assert "features.driver_stats" not in projection + + def test_versioned_read_of_requested_features(self): + projection = self._projection( + _make_feature_view(version_number=2), + _make_config(versioning=True), + requested=["trips_today"], + ) + assert "features.driver_stats_v2.trips_today" in projection + + +class TestConvertRawDocsVersioning: + """The converter reads back out of the namespace it was told about.""" + + def test_reads_versioned_namespace(self): + fv = _make_feature_view(version_number=2) + ts = datetime(2024, 1, 1, tzinfo=timezone.utc) + docs = { + b"e1": { + "features": { + "driver_stats": {"trips_today": 1}, + "driver_stats_v2": {"trips_today": 99}, + }, + "event_timestamps": {"driver_stats": ts, "driver_stats_v2": ts}, + } + } + + results = MongoDBOnlineStore._convert_raw_docs_to_proto( + [b"e1"], docs, fv, "driver_stats_v2" + ) + + assert results[0][1]["trips_today"].int64_val == 99 + + def test_defaults_to_table_name(self): + """Omitting fv_name keeps the pre-versioning behaviour for existing callers.""" + fv = _make_feature_view() + ts = datetime(2024, 1, 1, tzinfo=timezone.utc) + docs = { + b"e1": { + "features": {"driver_stats": {"trips_today": 5}}, + "event_timestamps": {"driver_stats": ts}, + } + } + + results = MongoDBOnlineStore._convert_raw_docs_to_proto([b"e1"], docs, fv) + + assert results[0][1]["trips_today"].int64_val == 5 + + def test_missing_version_namespace_reads_as_absent(self): + """A versioned read of data written before that version finds nothing.""" + fv = _make_feature_view(version_number=2) + ts = datetime(2024, 1, 1, tzinfo=timezone.utc) + docs = { + b"e1": { + "features": {"driver_stats": {"trips_today": 1}}, + "event_timestamps": {"driver_stats": ts}, + } + } + + results = MongoDBOnlineStore._convert_raw_docs_to_proto( + [b"e1"], docs, fv, "driver_stats_v2" + ) + + assert results[0] == (None, None) + + +class TestUpdateVersioning: + """Deleting a versioned feature view unsets only that version namespace.""" + + @staticmethod + def _unset(fv, config): + store = MongoDBOnlineStore() + collection = MagicMock() + store._get_collection = MagicMock(return_value=collection) + store.update( + config=config, + tables_to_delete=[fv], + tables_to_keep=[], + entities_to_delete=[], + entities_to_keep=[], + partial=False, + ) + return collection.update_many.call_args.args[1]["$unset"] + + def test_versioned_delete_unsets_versioned_namespace(self): + unset = self._unset( + _make_feature_view(version_number=2), _make_config(versioning=True) + ) + assert "features.driver_stats_v2" in unset + assert "event_timestamps.driver_stats_v2" in unset + assert "features.driver_stats" not in unset + + def test_unversioned_delete_unsets_plain_namespace(self): + unset = self._unset(_make_feature_view(), _make_config(versioning=False)) + assert "features.driver_stats" in unset + + +class TestVectorIndexVersioning: + """Each version gets its own Atlas vector search index and path.""" + + @staticmethod + def _vector_fv(version_number=None): + entity = Entity( + name="doc_id", join_keys=["doc_id"], value_type=ValueType.STRING + ) + fv = FeatureView( + name="docs", + entities=[entity], + ttl=timedelta(days=1), + schema=[ + Field( + name="embedding", + dtype=Array(Float32), + vector_index=True, + vector_length=4, + ) + ], + source=FileSource(path="fake.parquet", timestamp_field="event_timestamp"), + ) + if version_number is not None: + fv.current_version_number = version_number + return fv + + @staticmethod + def _created_index(fv, enable_versioning): + store = MongoDBOnlineStore() + collection = MagicMock() + collection.name = "test_project_latest" + collection.database.list_collection_names.return_value = ["test_project_latest"] + + def _list_search_indexes(*args, **kwargs): + # No index exists yet for the "already exists?" check, but the + # readiness poll that follows creation must find it READY or it + # would spin until vector_index_wait_timeout. + if kwargs.get("name"): + return [{"name": kwargs["name"], "status": "READY"}] + return [] + + collection.list_search_indexes.side_effect = _list_search_indexes + online_config = _make_config(vector_enabled=True).online_store + + store._ensure_vector_indexes(collection, [fv], online_config, enable_versioning) + + return collection.create_search_index.call_args.kwargs["model"] + + def test_unversioned_index_name_and_path(self): + model = self._created_index(self._vector_fv(), False) + assert model.document["name"] == "docs__embedding__vs_index" + assert ( + model.document["definition"]["fields"][0]["path"] + == "features.docs.embedding" + ) + + def test_versioned_index_name_and_path(self): + model = self._created_index(self._vector_fv(version_number=2), True) + assert model.document["name"] == "docs_v2__embedding__vs_index" + assert ( + model.document["definition"]["fields"][0]["path"] + == "features.docs_v2.embedding" + ) + + def test_drop_targets_versioned_index(self): + store = MongoDBOnlineStore() + collection = MagicMock() + collection.list_search_indexes.return_value = [ + {"name": "docs_v2__embedding__vs_index"} + ] + + store._drop_vector_indexes_for_tables( + collection, [self._vector_fv(version_number=2)], True + ) + + collection.drop_search_index.assert_called_once_with( + "docs_v2__embedding__vs_index" + ) + + +class TestMongoDBVersionedReadSupport: + """MongoDBOnlineStore is registered as supporting versioned reads.""" + + def test_allowed_with_version_tag(self): + store = MongoDBOnlineStore() + fv = _make_feature_view() + fv.projection.version_tag = 2 + # Should not raise + store._check_versioned_read_support([(fv, ["trips_today"])]) + + def test_allowed_without_version_tag(self): + store = MongoDBOnlineStore() + store._check_versioned_read_support([(_make_feature_view(), ["trips_today"])]) From f1da3d6a90e980c4021d98804084e774dad4842b Mon Sep 17 00:00:00 2001 From: Pavel Alekseev Date: Sat, 22 Aug 2026 17:02:11 +0300 Subject: [PATCH 2/2] fix: List Milvus in the unsupported versioned read message Milvus has been in the allow-list consulted by OnlineStore._is_versioned_read_supported() since versioned reads were added to it, but was never added to the sentence VersionedOnlineReadNotSupported prints, which tells the user which stores do support them. The message therefore names a store set smaller than reality. Signed-off-by: Pavel Alekseev --- sdk/python/feast/errors.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/python/feast/errors.py b/sdk/python/feast/errors.py index 0f7fac4e6de..4a39c376b2a 100644 --- a/sdk/python/feast/errors.py +++ b/sdk/python/feast/errors.py @@ -142,7 +142,7 @@ class VersionedOnlineReadNotSupported(FeastError): def __init__(self, store_name: str, version: int): super().__init__( f"Versioned feature reads (@v{version}) are not yet supported by {store_name}. " - f"Currently only SQLite, PostgreSQL, MySQL, FAISS, Redis, DynamoDB, and MongoDB support version-qualified feature references. " + f"Currently only SQLite, PostgreSQL, MySQL, FAISS, Redis, DynamoDB, Milvus, and MongoDB support version-qualified feature references. " )