feat(index): add scalar inverted index support (#342)

Add scalar INVERTED index creation, persistence, recovery, and query/search
  filter acceleration. Extend index spec handling, sidecar file management,
  snapshot references, and gRPC index parameter translation to support scalar
  indexes alongside existing vector indexes.

  Also add scalar index tests, snapshot coverage, and a demo for indexed scalar
  filter queries.

Signed-off-by: junjie.jiang <junjie.jiang@zilliz.com>

Author: junjiejiangjjj

examples/m11_scalar_index_demo.py

@@ -0,0 +1,128 @@
+"""M11 demo — scalar INVERTED index.
+
+Shows scalar indexes on query() and search() filters:
+
+    insert → flush → create_index(age/category) → query/search with filters
+    → inspect .sidx sidecars → reopen → load → query again
+
+Run:
+    python examples/m11_scalar_index_demo.py
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import tempfile
+
+import numpy as np
+
+from milvus_lite import CollectionSchema, DataType, FieldSchema, MilvusLite
+
+
+def _schema() -> CollectionSchema:
+    return CollectionSchema(fields=[
+        FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),
+        FieldSchema(name="vec", dtype=DataType.FLOAT_VECTOR, dim=4),
+        FieldSchema(name="age", dtype=DataType.INT64, nullable=True),
+        FieldSchema(name="category", dtype=DataType.VARCHAR, nullable=True, max_length=64),
+        FieldSchema(name="active", dtype=DataType.BOOL, nullable=True),
+    ])
+
+
+def _rows() -> list[dict]:
+    categories = ["tech", "news", "blog", "ai", None]
+    rows = []
+    for i in range(20):
+        rows.append({
+            "id": i,
+            "vec": [float(i == j) for j in range(4)],
+            "age": None if i % 9 == 0 else 18 + i,
+            "category": categories[i % len(categories)],
+            "active": None if i % 7 == 0 else i % 2 == 0,
+        })
+    return rows
+
+
+def _ids(rows: list[dict]) -> list[int]:
+    return [int(row["id"]) for row in rows]
+
+
+def _index_files(data_dir: str) -> list[str]:
+    index_dir = os.path.join(data_dir, "collections", "docs", "partitions", "_default", "indexes")
+    if not os.path.isdir(index_dir):
+        return []
+    return sorted(os.listdir(index_dir))
+
+
+def main() -> None:
+    data_dir = tempfile.mkdtemp(prefix="milvus_lite_m11_scalar_index_")
+    print(f"data_dir = {data_dir}")
+
+    try:
+        with MilvusLite(data_dir) as db:
+            col = db.create_collection("docs", _schema())
+            col.insert(_rows())
+            col.flush()
+            print("\n[1] inserted and flushed 20 rows")
+
+            print("\n[2] create scalar INVERTED indexes")
+            col.create_index("age", {"index_type": "INVERTED"})
+            col.create_index("category", {"index_type": "INVERTED"})
+            print(f"   indexes = {col.list_indexes()}")
+            print(f"   age index info = {col.get_index_info('age')}")
+            print(f"   category index info = {col.get_index_info('category')}")
+
+            print("\n[3] query() uses indexed scalar predicates")
+            adults = col.query("age >= 25 and age < 32", output_fields=["age", "category"])
+            print(f"   age >= 25 and age < 32 ids = {_ids(adults)}")
+            assert _ids(adults) == [7, 8, 10, 11, 12, 13]
+
+            tech_or_ai = col.query(
+                "category in ['tech', 'ai']",
+                output_fields=["category"],
+                limit=6,
+            )
+            print(f"   category in ['tech', 'ai'] first ids = {_ids(tech_or_ai)}")
+            assert all(row["category"] in {"tech", "ai"} for row in tech_or_ai)
+
+            print("\n[4] search() combines vector ranking with indexed scalar filter")
+            query = np.asarray([[1, 0, 0, 0]], dtype=np.float32).tolist()
+            results = col.search(
+                query,
+                top_k=5,
+                metric_type="L2",
+                expr="category == 'tech' and age is not null",
+                output_fields=["age", "category"],
+            )
+            for hit in results[0]:
+                entity = hit["entity"]
+                print(
+                    f"   id={hit['id']:2d} dist={hit['distance']:.1f} "
+                    f"age={entity['age']} category={entity['category']}"
+                )
+                assert entity["category"] == "tech"
+                assert entity["age"] is not None
+
+            print("\n[5] .sidx sidecar files were written")
+            sidecars = _index_files(data_dir)
+            for filename in sidecars:
+                print(f"   {filename}")
+            assert any(filename.endswith(".age.inverted.sidx") for filename in sidecars)
+            assert any(filename.endswith(".category.inverted.sidx") for filename in sidecars)
+
+        print("\n[6] reopen collection and load existing scalar index sidecars")
+        with MilvusLite(data_dir) as db:
+            reopened = db.get_collection("docs")
+            reopened.load()
+            rows = reopened.query("category == 'news'", output_fields=["category"])
+            print(f"   category == 'news' ids after reopen = {_ids(rows)}")
+            assert _ids(rows) == [1, 6, 11, 16]
+
+        print("\nOK — M11 demo passed: scalar INVERTED indexes work end-to-end")
+    finally:
+        shutil.rmtree(data_dir, ignore_errors=True)
+
+
+if __name__ == "__main__":
+    main()

milvus_lite/adapter/grpc/servicer.py

@@ -493,11 +493,17 @@ def Search(self, request, context):
             from milvus_lite.function.types import ID_FIELD, SCORE_FIELD
 
             col = self._db.get_collection(request.collection_name)
-            # Pull the canonical metric from the first IndexSpec if any.
-            first_spec = col._index_specs.get(col._vector_name) if col._index_specs else None  # noqa: SLF001
-            if first_spec is None and col._index_specs:
-                first_spec = next(iter(col._index_specs.values()))
-            default_metric = first_spec.metric_type if first_spec else "COSINE"
+            # Pull the canonical metric from a vector index if any.
+            first_spec = (
+                col._index_specs.get(col._vector_name)  # noqa: SLF001
+                if col._index_specs and col._vector_name is not None  # noqa: SLF001
+                else None
+            )
+            default_metric = (
+                first_spec.metric_type
+                if first_spec is not None and first_spec.metric_type != "NONE"
+                else "COSINE"
+            )
             parsed = parse_search_request(request, default_metric_type=default_metric)
 
             group_by_field = parsed.get("group_by_field")
@@ -645,9 +651,7 @@ def CreateIndex(self, request, context):
         """
         try:
             col = self._db.get_collection(request.collection_name)
-            params = kv_pairs_to_index_params_dict(
-                request.extra_params, field_name=request.field_name
-            )
+            params = kv_pairs_to_index_params_dict(request.extra_params)
             col.create_index(request.field_name, params)
             return common_pb2.Status(**success_status_kwargs())
         except MilvusLiteError as e:

milvus_lite/adapter/grpc/translators/index.py

@@ -28,40 +28,28 @@
 from milvus_lite.index.spec import IndexSpec
 
 
-def kv_pairs_to_index_params_dict(
-    extra_params: Iterable,
-    field_name: str = "",
-) -> dict:
+def kv_pairs_to_index_params_dict(extra_params: Iterable) -> dict:
     """Decode the CreateIndexRequest.extra_params KeyValuePair list
     into a dict suitable for ``Collection.create_index(field, params)``.
 
     Args:
         extra_params: iterable of common_pb2.KeyValuePair messages
-        field_name: fallback field_name from the outer request, used
-            if the KV list doesn't carry one
 
     Returns:
         ``{"index_type": ..., "metric_type": ..., "params": {...},
-            "search_params": {...}}``  — the shape Collection.create_index
-        consumes.
+            "search_params": {...}}``  — the index_params shape
+        Collection.create_index consumes.
 
     Raises:
-        SchemaValidationError: required keys (index_type, metric_type)
-            missing or malformed
+        SchemaValidationError: required keys are missing or malformed
     """
     kv: dict[str, str] = {p.key: p.value for p in extra_params}
 
     index_type = kv.get("index_type")
-    metric_type = kv.get("metric_type")
-
     if not index_type:
         raise SchemaValidationError(
             "create_index missing required 'index_type' parameter"
         )
-    if not metric_type:
-        raise SchemaValidationError(
-            "create_index missing required 'metric_type' parameter"
-        )
 
     # `params` is a JSON-encoded string in the wire format. pymilvus
     # uses utils.dumps which is plain json.dumps for dicts.
@@ -87,12 +75,14 @@ def kv_pairs_to_index_params_dict(
     else:
         search_params = {}
 
-    return {
+    out = {
         "index_type": index_type,
-        "metric_type": metric_type,
         "params": build_params,
         "search_params": search_params,
     }
+    if "metric_type" in kv:
+        out["metric_type"] = kv["metric_type"]
+    return out
 
 
 def index_spec_to_kv_pairs(spec: IndexSpec) -> List: