fix(vector-store): validate collection schema in MilvusStore.get_collection (#1344)

`get_collection()` attached any collection right after the existence check, with no look at its schema. A collection with an INT64 primary key, or one missing the `metadata` field entirely, would attach without complaint and only fail later, inside `get_vector()` or `get_metadata()`, with an error that gave no hint the real problem was upstream at attach time.

This adds a schema check between the attach and the assignment to `self.collection`, so a mismatch is caught at the point of failure instead of surfacing three calls later as an unrelated-looking error. The check validates against exactly the shape `create_collection()` builds: a `VARCHAR` primary key named `id` with `auto_id=False`, a `FLOAT_VECTOR` field named `vector`, and a `JSON` field named `metadata`. Anything else, wrong dtype, wrong name, a missing field, or an auto-generated id, is rejected before the store ever holds a reference to it.

The auto_id and metadata-dtype checks were added in a second pass after review. A collection with `auto_id=True` still attached cleanly and only broke once the store tried to insert with the explicit ids it always sends, and a `metadata` field that existed but wasn't `JSON`-typed only broke during a later write or metadata filter, for the same reason: schema drift that looked fine at attach time and failed downstream instead of at the source.

Nine tests cover this: the one matching-schema case that should succeed, and each rejection path independently, wrong pk dtype, missing pk, wrong pk name, auto_id pk, missing vector field, wrong vector dtype, missing metadata field, and non-JSON metadata.

Closes #1331.

semantica/vector_store/milvus_store.py

@@ -438,6 +438,46 @@ class MilvusStore:
                 raise ProcessingError(f"Collection {collection_name} does not exist")
 
             collection = Collection(collection_name)
+            # Reject schemas that don't match create_collection()'s shape:
+            # id/VARCHAR pk + vector + metadata. Otherwise an incompatible
+            # collection attaches and fails far later in get_vector/get_metadata.
+            schema = getattr(collection, "schema", None)
+            fields = list(getattr(schema, "fields", None) or [])
+            pk = [f for f in fields if getattr(f, "is_primary", False)]
+            if (
+                not pk
+                or pk[0].name != "id"
+                or getattr(getattr(pk[0], "dtype", None), "name", None) != "VARCHAR"
+                or getattr(pk[0], "auto_id", False)
+            ):
+                raise ProcessingError(
+                    f"Collection '{collection_name}' has an invalid primary key: "
+                    "expected VARCHAR field 'id' without auto_id"
+                )
+            vector_field = next((f for f in fields if f.name == "vector"), None)
+            if vector_field is None:
+                raise ProcessingError(
+                    f"Collection '{collection_name}' is missing required field 'vector'"
+                )
+            if (
+                getattr(getattr(vector_field, "dtype", None), "name", None)
+                != "FLOAT_VECTOR"
+            ):
+                raise ProcessingError(
+                    f"Collection '{collection_name}' has an invalid vector field: "
+                    "expected FLOAT_VECTOR 'vector'"
+                )
+            metadata_field = next((f for f in fields if f.name == "metadata"), None)
+            if metadata_field is None:
+                raise ProcessingError(
+                    f"Collection '{collection_name}' is missing required field 'metadata'"
+                )
+            if getattr(getattr(metadata_field, "dtype", None), "name", None) != "JSON":
+                raise ProcessingError(
+                    f"Collection '{collection_name}' has an invalid metadata field: "
+                    "expected JSON 'metadata'"
+                )
+
             self.collection = MilvusCollection(collection, collection_name)
             self.search_engine = MilvusSearch(self.collection)
             return self.collection

tests/vector_store/test_milvus_get_collection.py

@@ -0,0 +1,141 @@
+"""Tests for MilvusStore.get_collection schema validation (#1331)."""
+
+from unittest import TestCase
+from unittest.mock import MagicMock, patch
+
+from semantica.vector_store.milvus_store import MilvusStore
+from semantica.utils.exceptions import ProcessingError
+
+
+def _field(name, dtype_name, primary=False, auto_id=False):
+    f = MagicMock()
+    f.name = name
+    f.is_primary = primary
+    f.auto_id = auto_id
+    f.dtype.name = dtype_name
+    return f
+
+
+class MilvusGetCollectionSchemaTest(TestCase):
+    def setUp(self):
+        self.patches = [
+            patch("semantica.vector_store.milvus_store.MILVUS_AVAILABLE", True),
+            patch("semantica.vector_store.milvus_store.utility"),
+            patch("semantica.vector_store.milvus_store.Collection"),
+        ]
+        for p in self.patches:
+            p.start()
+        # utility.has_collection() must return truthy
+        import semantica.vector_store.milvus_store as m
+
+        m.utility.has_collection.return_value = True
+
+    def tearDown(self):
+        for p in reversed(self.patches):
+            p.stop()
+
+    def _make_store(self, coll):
+        store = MilvusStore()
+        store.client = MagicMock()  # skip real connect
+        import semantica.vector_store.milvus_store as m
+
+        m.Collection.return_value = coll
+        return store
+
+    def _assert_rejected(self, store, expected_msg):
+        with self.assertRaises(ProcessingError) as ctx:
+            store.get_collection("c")
+        self.assertIn(expected_msg, str(ctx.exception))
+        self.assertIsNone(store.collection)
+
+    def test_accepts_matching_schema(self):
+        coll = MagicMock()
+        coll.schema.fields = [
+            _field("id", "VARCHAR", primary=True),
+            _field("vector", "FLOAT_VECTOR"),
+            _field("metadata", "JSON"),
+        ]
+        store = self._make_store(coll)
+        result = store.get_collection("c")
+        self.assertIsNotNone(result)
+        self.assertIsNotNone(store.collection)
+        self.assertIsNotNone(store.search_engine)
+        self.assertEqual(store.collection.collection_name, "c")
+
+    def test_rejects_non_varchar_primary_key(self):
+        coll = MagicMock()
+        coll.schema.fields = [
+            _field("id", "INT64", primary=True),
+            _field("vector", "FLOAT_VECTOR"),
+            _field("metadata", "JSON"),
+        ]
+        store = self._make_store(coll)
+        self._assert_rejected(store, "has an invalid primary key")
+
+    def test_rejects_missing_primary_key(self):
+        coll = MagicMock()
+        coll.schema.fields = [
+            _field("id", "VARCHAR"),
+            _field("vector", "FLOAT_VECTOR"),
+            _field("metadata", "JSON"),
+        ]
+        store = self._make_store(coll)
+        self._assert_rejected(store, "has an invalid primary key")
+
+    def test_rejects_wrongly_named_primary_key(self):
+        coll = MagicMock()
+        coll.schema.fields = [
+            _field("pk", "VARCHAR", primary=True),
+            _field("vector", "FLOAT_VECTOR"),
+            _field("metadata", "JSON"),
+        ]
+        store = self._make_store(coll)
+        self._assert_rejected(store, "has an invalid primary key")
+
+    def test_rejects_missing_metadata_field(self):
+        coll = MagicMock()
+        coll.schema.fields = [
+            _field("id", "VARCHAR", primary=True),
+            _field("vector", "FLOAT_VECTOR"),
+        ]
+        store = self._make_store(coll)
+        self._assert_rejected(store, "is missing required field 'metadata'")
+
+    def test_rejects_missing_vector_field(self):
+        coll = MagicMock()
+        coll.schema.fields = [
+            _field("id", "VARCHAR", primary=True),
+            _field("metadata", "JSON"),
+        ]
+        store = self._make_store(coll)
+        self._assert_rejected(store, "is missing required field 'vector'")
+
+    def test_rejects_wrong_vector_dtype(self):
+        coll = MagicMock()
+        coll.schema.fields = [
+            _field("id", "VARCHAR", primary=True),
+            _field("vector", "BINARY_VECTOR"),
+            _field("metadata", "JSON"),
+        ]
+        store = self._make_store(coll)
+        self._assert_rejected(store, "has an invalid vector field")
+
+    def test_rejects_auto_id_primary_key(self):
+        coll = MagicMock()
+        coll.schema.fields = [
+            _field("id", "VARCHAR", primary=True, auto_id=True),
+            _field("vector", "FLOAT_VECTOR"),
+            _field("metadata", "JSON"),
+        ]
+        store = self._make_store(coll)
+        self._assert_rejected(store, "has an invalid primary key")
+
+    def test_rejects_non_json_metadata(self):
+        coll = MagicMock()
+        coll.schema.fields = [
+            _field("id", "VARCHAR", primary=True),
+            _field("vector", "FLOAT_VECTOR"),
+            _field("metadata", "STRING"),
+        ]
+        store = self._make_store(coll)
+        self._assert_rejected(store, "has an invalid metadata field")