Merge branch 'master' into feat/redis-vector-search

falloficaruss · web-flow · commit b660dc9ddfa4 · 2026-06-10T08:16:08.000+05:30
diff --git a/sdk/python/feast/infra/online_stores/milvus_online_store/milvus.py b/sdk/python/feast/infra/online_stores/milvus_online_store/milvus.py
@@ -4,7 +4,7 @@
 from pathlib import Path
 from typing import Any, Callable, Dict, List, Literal, Optional, Sequence, Tuple, Union
 
-from pydantic import StrictStr
+from pydantic import StrictStr, field_validator
 from pymilvus import (
     CollectionSchema,
     DataType,
@@ -115,6 +115,14 @@ class MilvusOnlineStoreConfig(FeastConfigBaseModel, VectorStoreConfig):
     nlist: Optional[int] = 128
     username: Optional[StrictStr] = ""
     password: Optional[StrictStr] = ""
+    varchar_max_length: Optional[int] = 65535
+
+    @field_validator("varchar_max_length")
+    @classmethod
+    def validate_varchar_max_length(cls, v):
+        if v is not None and not (1 <= v <= 65535):
+            raise ValueError(f"varchar_max_length must be between 1 and 65535, got {v}")
+        return v
 
 
 class MilvusOnlineStore(OnlineStore):
@@ -171,12 +179,12 @@ def _get_or_create_collection(
         if collection_name not in self._collections:
             # Create a composite key by combining entity fields
             composite_key_name = _get_composite_key_name(table)
-
+            varchar_max_length = int(config.online_store.varchar_max_length or 65535)
             fields = [
                 FieldSchema(
                     name=composite_key_name,
                     dtype=DataType.VARCHAR,
-                    max_length=512,
+                    max_length=varchar_max_length,
                     is_primary=True,
                 ),
                 FieldSchema(name="event_ts", dtype=DataType.INT64),
@@ -203,14 +211,28 @@ def _get_or_create_collection(
                             )
                         )
                     else:
+                        if "max_length" in field.tags:
+                            try:
+                                field_max_length = int(field.tags["max_length"])
+                            except (ValueError, TypeError):
+                                raise ValueError(
+                                    f"Field '{field.name}' has invalid max_length tag"
+                                    f" '{field.tags['max_length']}': must be an integer."
+                                )
+                            if not (1 <= field_max_length <= 65535):
+                                raise ValueError(
+                                    f"Field '{field.name}' max_length tag must be"
+                                    f" between 1 and 65535, got {field_max_length}"
+                                )
+                        else:
+                            field_max_length = varchar_max_length
                         fields.append(
                             FieldSchema(
                                 name=field.name,
                                 dtype=DataType.VARCHAR,
-                                max_length=512,
+                                max_length=field_max_length,
                             )
                         )
-
             schema = CollectionSchema(
                 fields=fields, description="Feast feature view data"
             )
diff --git a/sdk/python/tests/unit/infra/online_store/test_milvus_varchar_max_length.py b/sdk/python/tests/unit/infra/online_store/test_milvus_varchar_max_length.py
@@ -0,0 +1,66 @@
+"""Unit test for Milvus varchar_max_length configuration."""
+
+from datetime import timedelta
+from unittest.mock import MagicMock, patch
+
+import pytest
+from pydantic import ValidationError
+
+from feast import Entity, FeatureView
+from feast.field import Field
+from feast.infra.online_stores.milvus_online_store.milvus import (
+    MilvusOnlineStore,
+    MilvusOnlineStoreConfig,
+)
+from feast.types import Float32, String
+from feast.value_type import ValueType
+
+
+@patch("feast.infra.online_stores.milvus_online_store.milvus.MilvusClient")
+def test_varchar_max_length(mock_client_cls):
+    # -- config: default and custom values ------------------------------------
+    assert MilvusOnlineStoreConfig().varchar_max_length == 65535
+    assert MilvusOnlineStoreConfig(varchar_max_length=1024).varchar_max_length == 1024
+
+    # -- config: out-of-bounds values raise ValidationError -------------------
+    for bad in (0, -1, 65536):
+        with pytest.raises(ValidationError):
+            MilvusOnlineStoreConfig(varchar_max_length=bad)
+
+    # -- schema: configured value reaches every VARCHAR FieldSchema -----------
+    mock_client = MagicMock()
+    mock_client_cls.return_value = mock_client
+    mock_client.has_collection.return_value = False
+
+    entity = Entity(
+        name="driver_id", join_keys=["driver_id"], value_type=ValueType.INT64
+    )
+    fv = FeatureView(
+        name="driver_stats",
+        entities=[entity],
+        ttl=timedelta(days=1),
+        schema=[
+            Field(name="trips_today", dtype=Float32),
+            Field(name="wiki_summary", dtype=String),
+        ],
+    )
+
+    config = MagicMock()
+    config.project = "test_project"
+    config.entity_key_serialization_version = 2
+    config.registry.enable_online_feature_view_versioning = False
+    config.provider = "local"
+    config.repo_path = None
+    config.online_store = MilvusOnlineStoreConfig(varchar_max_length=4096)
+
+    store = MilvusOnlineStore()
+    store._collections = {}
+    store.client = mock_client
+    store._get_or_create_collection(config, fv)
+
+    schema = mock_client.create_collection.call_args.kwargs["schema"]
+    for field in schema.fields:
+        if hasattr(field, "max_length") and field.max_length is not None:
+            assert field.max_length == 4096, (
+                f"field '{field.name}': got {field.max_length}"
+            )
