@@ -97,6 +97,8 @@ online_store:
 | `consistency_level` | unset | `Strong`, `Bounded`, `Session` or `Eventually`. Sent with every read and search. When unset, Milvus uses the collection's level. |
 | `collection_consistency_level` | unset | `Strong`, `Bounded`, `Session` or `Eventually`. Set when Feast creates a collection. When unset, Milvus uses its default (`Bounded`). |
 | `partition_key` | unset | Field to use as the Milvus partition key in feature views that contain it. See [Partition key](#partition-key). |
+| `num_partitions` | unset | Number of partitions of a new collection that has a partition key. When unset, Milvus uses its default. |
+| `partition_key_isolation` | unset | Enables partition key isolation on a new collection that has a partition key. When unset, Milvus uses its default. |
 | `vector_enabled` | `true` | Enables vector search. |
 | `varchar_max_length` | `65535` | Default `max_length` of VARCHAR fields. Override per field with the `max_length` tag. |
 | `enable_openai_compatible_store` | `false` | Store numeric features as native Milvus numeric types. |
@@ -168,6 +170,12 @@ store.retrieve_online_documents_v2(
 )
 ```
 
+`num_partitions` sets how many partitions a new collection with a partition key gets, and
+`partition_key_isolation` enables
+[partition key isolation](https://milvus.io/docs/use-partition-key.md#Use-Partition-Key-Isolation)
+for it, which Milvus supports with the HNSW index. Both settings are ignored for feature views
+without a partition key, and Milvus Lite ignores `num_partitions`.
+
 The partition key field must be stored as `VARCHAR` or `INT64`. String fields are always `VARCHAR`,
 and integer fields are `VARCHAR` unless native numeric types are enabled, in which case `Int64` works
 and `Int32` does not.
 
@@ -250,11 +250,24 @@ class MilvusOnlineStoreConfig(FeastConfigBaseModel, VectorStoreConfig):
     # Field to use as the Milvus partition key in feature views that contain it.
     # A feature view's "milvus.partition_key" tag takes precedence.
     partition_key: Optional[StrictStr] = None
+    # Number of partitions of a new collection that has a partition key. When
+    # unset, Milvus uses its default.
+    num_partitions: Optional[int] = None
+    # Partition key isolation of a new collection that has a partition key.
+    # When unset, Milvus uses its default.
+    partition_key_isolation: Optional[bool] = None
     username: Optional[StrictStr] = ""
     password: Optional[StrictStr] = ""
     enable_openai_compatible_store: Optional[bool] = False
     varchar_max_length: Optional[int] = 65535
 
+    @field_validator("num_partitions")
+    @classmethod
+    def validate_num_partitions(cls, v):
+        if v is not None and v < 1:
+            raise ValueError(f"num_partitions must be at least 1, got {v}")
+        return v
+
     @field_validator("varchar_max_length")
     @classmethod
     def validate_varchar_max_length(cls, v):
@@ -445,6 +458,7 @@ def _get_or_create_collection(
                     schema=schema,
                     index_params=index_params,
                     **_collection_consistency_kwargs(config.online_store),
+                    **_partition_kwargs(config.online_store, partition_key),
                 )
             else:
                 self._ensure_loaded(collection_name)
@@ -1195,6 +1209,22 @@ def _consistency_kwargs(online_config: MilvusOnlineStoreConfig) -> Dict[str, Any
     return {}
 
 
+def _partition_kwargs(
+    online_config: MilvusOnlineStoreConfig, partition_key: Optional[str]
+) -> Dict[str, Any]:
+    """Partition settings only apply to a collection that has a partition key."""
+    kwargs: Dict[str, Any] = {}
+    if not partition_key:
+        return kwargs
+    if online_config.num_partitions is not None:
+        kwargs["num_partitions"] = online_config.num_partitions
+    if online_config.partition_key_isolation is not None:
+        kwargs["properties"] = {
+            "partitionkey.isolation": str(online_config.partition_key_isolation).lower()
+        }
+    return kwargs
+
+
 def _collection_consistency_kwargs(
     online_config: MilvusOnlineStoreConfig,
 ) -> Dict[str, Any]:
 
@@ -409,6 +409,63 @@ def test_partition_key_filtering(
     assert [hit["city"].string_val for hit in hits] == ["Rome"]
 
 
+def test_partition_options(
+    tmp_path: Path, project: str, store: MilvusOnlineStore
+) -> None:
+    config = _repo_config(
+        tmp_path,
+        project,
+        partition_key="brand_id",
+        num_partitions=4,
+        partition_key_isolation=True,
+        index_type="HNSW",
+        consistency_level="Strong",
+    )
+    fv = FeatureView(
+        name="products",
+        entities=[
+            Entity(
+                name="driver_id", join_keys=["driver_id"], value_type=ValueType.INT64
+            )
+        ],
+        ttl=timedelta(days=1),
+        schema=[
+            Field(name="driver_id", dtype=Int64),
+            Field(name="brand_id", dtype=String),
+            Field(
+                name="embedding",
+                dtype=Array(Float32),
+                vector_index=True,
+                vector_search_metric="COSINE",
+            ),
+            Field(name="city", dtype=String),
+        ],
+    )
+    store.update(config, [], [fv], [], [], partial=False)
+
+    assert store.client is not None
+    description = store.client.describe_collection(f"{project}_{fv.name}")
+    assert description["num_partitions"] == 4
+    properties = description["properties"]
+    if isinstance(properties, list):
+        properties = {entry["key"]: entry["value"] for entry in properties}
+    assert properties["partitionkey.isolation"] == "true"
+
+    rows = _vector_rows()
+    rows[1]["brand_id"] = ValueProto(string_val="acme")
+    rows[2]["brand_id"] = ValueProto(string_val="globex")
+    _write_rows(store, config, fv, rows)
+    hits = _search(
+        store,
+        config,
+        fv,
+        [1.0, 0.0],
+        top_k=5,
+        filters=ComparisonFilter(type="eq", key="brand_id", value="globex"),
+    )
+    assert [hit["city"].string_val for hit in hits] == ["Rome"]
+
+
 def test_online_delete(tmp_path: Path, project: str, store: MilvusOnlineStore) -> None:
     config = _repo_config(tmp_path, project, consistency_level="Strong")
     fv = _vector_feature_view()
 
@@ -748,6 +748,43 @@ def test_warns_when_existing_collection_lacks_partition_key(
     mock_client.create_collection.assert_not_called()
 
 
+@patch(f"{MILVUS_MODULE}.MilvusClient")
+def test_partition_options_sent_only_for_partitioned_collections(
+    mock_client_cls: MagicMock,
+) -> None:
+    mock_client = _mock_client(mock_client_cls, has_collection=False)
+    config = _mock_config(
+        partition_key="brand_id", num_partitions=32, partition_key_isolation=True
+    )
+    store = MilvusOnlineStore()
+
+    store._get_or_create_collection(config, _catalog_feature_view())
+    kwargs = mock_client.create_collection.call_args.kwargs
+    assert kwargs["num_partitions"] == 32
+    assert kwargs["properties"] == {"partitionkey.isolation": "true"}
+
+    store._get_or_create_collection(config, _scalar_feature_view())
+    kwargs = mock_client.create_collection.call_args.kwargs
+    assert "num_partitions" not in kwargs and "properties" not in kwargs
+
+
+@patch(f"{MILVUS_MODULE}.MilvusClient")
+def test_partition_options_not_sent_when_unset(mock_client_cls: MagicMock) -> None:
+    mock_client = _mock_client(mock_client_cls, has_collection=False)
+
+    MilvusOnlineStore()._get_or_create_collection(
+        _mock_config(partition_key="brand_id"), _catalog_feature_view()
+    )
+
+    kwargs = mock_client.create_collection.call_args.kwargs
+    assert "num_partitions" not in kwargs and "properties" not in kwargs
+
+
+def test_invalid_num_partitions_rejected() -> None:
+    with pytest.raises(ValidationError, match="num_partitions"):
+        MilvusOnlineStoreConfig(num_partitions=0)
+
+
 def test_online_delete_by_entity_key_and_by_filter(tmp_path: Path) -> None:
     config = _lite_config(tmp_path)
     fv = _scalar_feature_view()