diff --git a/sdk/python/feast/errors.py b/sdk/python/feast/errors.py
index 515a6c39b11..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, and DynamoDB support version-qualified feature references. "
+            f"Currently only SQLite, PostgreSQL, MySQL, FAISS, Redis, DynamoDB, Milvus, 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"])])