""" Large TopK Feature E2E Tests Feature: collection-level property `query_mode: large_topk` - Backend index auto-switches to IVF RBQ2 - Supports topk up to 1M level - alter/drop property requires dropping vector index first Test Plan: tests/python_client/docs/test-plan-large-topk.md Issue: https://github.com/milvus-io/milvus/issues/48725 """ import pytest from base.client_v2_base import TestMilvusClientV2Base from common import common_func as cf from common import common_type as ct from common.common_type import CaseLabel, CheckTasks from pymilvus import AnnSearchRequest, DataType, MilvusException, RRFRanker prefix = "large_topk" default_nb = 3000 # > 1024 to trigger IVF index build default_dim = ct.default_dim # 128 default_nq = ct.default_nq # 2 default_limit = ct.default_limit # 10 vec_field = ct.default_float_vec_field_name # "float_vector" large_topk_first = 16385 # first topk above the normal 16384 limit large_topk_total = 21000 # total rows in col_large_topk (> large_topk_first + default_nb for headroom) large_topk_max = 1_000_000 # maximum supported topk in large_topk mode @pytest.mark.xdist_group("TestLargeTopkShared") class TestLargeTopkShared(TestMilvusClientV2Base): """ L0 + L1 read-only tests. Two shared collections are prepared once: - col_large_topk: query_mode=large_topk, 6×default_nb vectors, FLAT index - col_normal: no query_mode set, default_nb vectors, FLAT index All tests are read-only; no data modification or index changes. """ def setup_class(self): super().setup_class(self) self.col_large_topk = "TestLargeTopkSharedLargeTopk" + cf.gen_unique_str("_") self.col_normal = "TestLargeTopkSharedNormal" + cf.gen_unique_str("_") @pytest.fixture(scope="class", autouse=True) def prepare_collections(self, request): client = self._client() def _create(col_name, enable_large_topk, batches=1): schema = self.create_schema(client)[0] schema.add_field("id", DataType.INT64, is_primary=True, auto_id=True) schema.add_field(vec_field, DataType.FLOAT_VECTOR, dim=default_dim) query_mode_props = {"query_mode": "large_topk"} if enable_large_topk else None self.create_collection(client, col_name, schema=schema, properties=query_mode_props, force_teardown=False) index_params = self.prepare_index_params(client)[0] # FLAT: 100% recall, simplifies assertions index_params.add_index(vec_field, index_type="FLAT", metric_type="L2") self.create_index(client, col_name, index_params) self.load_collection(client, col_name) rows = [{vec_field: cf.gen_vectors(1, default_dim)[0]} for _ in range(default_nb)] for _ in range(batches): self.insert(client, col_name, rows) self.flush(client, col_name) # large_topk_total rows total; topk tests use large_topk_total - default_nb for ~1 batch headroom _create(self.col_large_topk, enable_large_topk=True, batches=large_topk_total // default_nb) _create(self.col_normal, enable_large_topk=False) def teardown(): client.drop_collection(self.col_large_topk) client.drop_collection(self.col_normal) request.addfinalizer(teardown) @pytest.mark.tags(CaseLabel.L0) def test_create_with_large_topk_property(self): """ target: verify query_mode=large_topk property is correctly set at create time method: 1. describe_collection and check properties dict contains query_mode=large_topk 2. search with limit=100, check nq/limit/metric expected: property present; search returns 100 results with ascending L2 distances """ client = self._client() desc = client.describe_collection(self.col_large_topk) props = desc.get("properties", {}) assert props.get("query_mode") == "large_topk", f"property not set: {props}" vectors = cf.gen_vectors(default_nq, default_dim) self.search(client, self.col_large_topk, data=vectors, anns_field=vec_field, limit=100, check_task=CheckTasks.check_search_results, check_items={"enable_milvus_client_api": True, "nq": default_nq, "limit": 100, "metric": "L2"}) @pytest.mark.tags(CaseLabel.L1) @pytest.mark.parametrize("topk", [1, 100, 16384]) def test_search_various_topk(self, topk): """ target: verify search with various topk values all return correct counts method: search col_with_prop with topk in [1, 100, 16384], check nq/limit/metric expected: each search returns exactly min(topk, default_nb) hits per query """ client = self._client() vectors = cf.gen_vectors(default_nq, default_dim) expected_limit = min(topk, large_topk_total) self.search(client, self.col_large_topk, data=vectors, anns_field=vec_field, limit=topk, check_task=CheckTasks.check_search_results, check_items={"enable_milvus_client_api": True, "nq": default_nq, "limit": expected_limit, "metric": "L2"}) @pytest.mark.tags(CaseLabel.L1) def test_search_result_consistency(self): """ target: verify repeated searches return identical results (IDs and distances) method: search same query vector twice, compare id list and distance list expected: ids and distances are identical across two calls """ client = self._client() vectors = cf.gen_vectors(1, default_dim) res1 = client.search(self.col_large_topk, data=vectors, limit=50, anns_field=vec_field) res2 = client.search(self.col_large_topk, data=vectors, limit=50, anns_field=vec_field) ids1 = [r["id"] for r in res1[0]] ids2 = [r["id"] for r in res2[0]] dist1 = [r["distance"] for r in res1[0]] dist2 = [r["distance"] for r in res2[0]] assert ids1 == ids2, f"Inconsistent IDs: {ids1[:5]} vs {ids2[:5]}" assert dist1 == dist2, f"Inconsistent distances: {dist1[:5]} vs {dist2[:5]}" @pytest.mark.tags(CaseLabel.L0) @pytest.mark.parametrize("topk", [large_topk_first, large_topk_total - default_nb]) def test_large_topk_above_normal_limit(self, topk): """ target: verify query_mode=large_topk allows topk above 16384 without error (core MVP) method: search col_large_topk (large_topk_total vectors) with topk in [large_topk_first, large_topk_total - default_nb] expected: search completes without exception, returns results with ascending L2 distances. Note: exact result count is not asserted — large_topk forces IVF index which does not guarantee 100% recall, so returned count may be < topk. """ client = self._client() results = client.search(self.col_large_topk, data=cf.gen_vectors(default_nq, default_dim), anns_field=vec_field, limit=topk) for hits in results: assert len(hits) > 0, "Expected non-empty results" distances = [h["distance"] for h in hits] assert distances == sorted(distances), "L2 distances should be ascending" @pytest.mark.tags(CaseLabel.L2) def test_topk_without_large_topk_property(self): """ target: verify topk>16384 is rejected when query_mode=large_topk is not set method: 1. search col_without_prop with limit=16384 — should succeed 2. search col_without_prop with limit=large_topk_first — should raise MilvusException expected: limit=16384 OK; limit=large_topk_first raises error with message about invalid topk """ client = self._client() vectors = cf.gen_vectors(default_nq, default_dim) # Normal topk limit works fine self.search(client, self.col_normal, data=vectors, anns_field=vec_field, limit=16384, check_task=CheckTasks.check_search_results, check_items={"enable_milvus_client_api": True, "nq": default_nq, "limit": default_nb, "metric": "L2"}) # Above limit must be rejected error = {ct.err_code: 65535, ct.err_msg: f"topk [{large_topk_first}] is invalid, it should be in range [1, 16384]"} self.search(client, self.col_normal, data=vectors, anns_field=vec_field, limit=large_topk_first, check_task=CheckTasks.err_res, check_items=error) # Note: search_iterator and query_iterator are NOT affected by query_mode=large_topk. # The SDK enforces batch_size <= 16384 client-side (ParamError, unrelated to large_topk). # The iterator `limit` (total result count) uses internal pagination with batch_size <= 16384, # so per-request topk never exceeds 16384. No large_topk-specific iterator tests are needed. @pytest.mark.tags(CaseLabel.L1) def test_query_large_limit(self): """ target: verify query() with limit > 16384 works when query_mode=large_topk is set method: query col_large_topk with limit=large_topk_first, verify results returned expected: returns large_topk_first results without error """ client = self._client() res = client.query( self.col_large_topk, filter="", output_fields=["id"], limit=large_topk_first, ) assert len(res) == large_topk_first, \ f"Expected {large_topk_first} results, got {len(res)}" @pytest.mark.tags(CaseLabel.L2) def test_query_without_property_fails(self): """ target: verify query() with limit > 16384 is rejected when query_mode=large_topk is NOT set method: query col_normal with limit=large_topk_first expected: MilvusException with invalid topk message """ client = self._client() with pytest.raises(MilvusException) as exc_info: client.query( self.col_normal, filter="", output_fields=["id"], limit=large_topk_first, ) assert str(large_topk_first) in str(exc_info.value), \ f"Expected topk error, got: {exc_info.value}" # --------------------------------------------------------------------------- # Independent Tests (each test owns its own collection) # --------------------------------------------------------------------------- class TestLargeTopkIndependent(TestMilvusClientV2Base): """ Tests that require modifying index or property — each test gets its own collection. force_teardown=True (default) ensures cleanup even on failure. """ def _setup_col(self, client, enable_large_topk=True, nb=default_nb): """Create collection with optional query_mode=large_topk, FLAT index, insert nb rows.""" col = cf.gen_collection_name_by_testcase_name(module_index=2) schema = self.create_schema(client)[0] schema.add_field("id", DataType.INT64, is_primary=True, auto_id=True) schema.add_field(vec_field, DataType.FLOAT_VECTOR, dim=default_dim) query_mode_props = {"query_mode": "large_topk"} if enable_large_topk else None self.create_collection(client, col, schema=schema, properties=query_mode_props, force_teardown=True) index_params = self.prepare_index_params(client)[0] index_params.add_index(vec_field, index_type="FLAT", metric_type="L2") self.create_index(client, col, index_params) self.load_collection(client, col) if nb > 0: rows = [{vec_field: cf.gen_vectors(1, default_dim)[0]} for _ in range(nb)] self.insert(client, col, rows) self.flush(client, col) return col @pytest.mark.tags(CaseLabel.L1) def test_alter_collection_add_property(self): """ target: verify alter_collection_properties correctly adds query_mode=large_topk method: 1. create collection without property, build FLAT index, insert data 2. release → drop_index → alter_collection_properties 3. describe_collection to verify property 4. rebuild index, load, search with limit=100 expected: property set; search returns 100 results with ascending L2 distances """ client = self._client() col = self._setup_col(client, enable_large_topk=False) self.release_collection(client, col) self.drop_index(client, col, vec_field) self.alter_collection_properties(client, col, properties={"query_mode": "large_topk"}) desc = client.describe_collection(col) assert desc.get("properties", {}).get("query_mode") == "large_topk" index_params = self.prepare_index_params(client)[0] index_params.add_index(vec_field, index_type="FLAT", metric_type="L2") self.create_index(client, col, index_params) self.load_collection(client, col) self.search(client, col, data=cf.gen_vectors(default_nq, default_dim), anns_field=vec_field, limit=100, check_task=CheckTasks.check_search_results, check_items={"enable_milvus_client_api": True, "nq": default_nq, "limit": 100, "metric": "L2"}) @pytest.mark.tags(CaseLabel.L1) def test_drop_collection_property(self): """ target: verify drop_collection_properties removes query_mode and restores normal behavior method: 1. create collection with property, build FLAT index, insert data 2. release → drop_index → drop_collection_properties 3. describe_collection to verify property absent 4. rebuild index, load, search with limit=default_limit expected: property absent; normal search returns default_limit results """ client = self._client() col = self._setup_col(client, enable_large_topk=True) self.release_collection(client, col) self.drop_index(client, col, vec_field) self.drop_collection_properties(client, col, property_keys=["query_mode"]) desc = client.describe_collection(col) assert "query_mode" not in desc.get("properties", {}), \ f"property still present: {desc.get('properties')}" index_params = self.prepare_index_params(client)[0] index_params.add_index(vec_field, index_type="FLAT", metric_type="L2") self.create_index(client, col, index_params) self.load_collection(client, col) self.search(client, col, data=cf.gen_vectors(default_nq, default_dim), anns_field=vec_field, limit=default_limit, check_task=CheckTasks.check_search_results, check_items={"enable_milvus_client_api": True, "nq": default_nq, "limit": default_limit, "metric": "L2"}) @pytest.mark.tags(CaseLabel.L1) def test_alter_property_without_dropping_index_fails(self): """ target: verify alter_collection_properties is rejected when vector index exists method: create collection with index, call alter_collection_properties directly expected: MilvusException with error code 702 and message containing "vector index" """ client = self._client() col = self._setup_col(client, enable_large_topk=False, nb=0) error = {ct.err_code: 702, ct.err_msg: "can not alter query_mode if the collection already has a vector index"} self.alter_collection_properties(client, col, properties={"query_mode": "large_topk"}, check_task=CheckTasks.err_res, check_items=error) @pytest.mark.tags(CaseLabel.L1) def test_drop_property_without_dropping_index_fails(self): """ target: verify drop_collection_properties is rejected when vector index exists method: create collection with large_topk property and index, call drop directly expected: MilvusException with error code 702 and message containing "vector index" """ client = self._client() col = self._setup_col(client, enable_large_topk=True, nb=0) error = {ct.err_code: 702, ct.err_msg: "can not alter query_mode if the collection already has a vector index"} self.drop_collection_properties(client, col, property_keys=["query_mode"], check_task=CheckTasks.err_res, check_items=error) @pytest.mark.tags(CaseLabel.L2) def test_empty_collection_search(self): """ target: verify search on empty large_topk collection returns 0 results method: create collection with property, build FLAT index, load, search without inserting expected: 0 results returned, no exception """ client = self._client() col = self._setup_col(client, enable_large_topk=True, nb=0) res = client.search(col, data=cf.gen_vectors(default_nq, default_dim), limit=default_limit, anns_field=vec_field) for hits in res: assert len(hits) == 0, f"Empty collection should return 0 results, got {len(hits)}" @pytest.mark.tags(CaseLabel.L2) def test_add_then_drop_property_roundtrip(self): """ target: verify adding then dropping query_mode property restores normal behavior method: 1. create plain collection, build FLAT index, insert data 2. drop index → alter_collection_properties (add) → rebuild index → search 3. drop index → drop_collection_properties → rebuild index → search 4. verify property absent after final drop expected: both searches return default_limit results; property absent after drop """ client = self._client() col = self._setup_col(client, enable_large_topk=False) vectors = cf.gen_vectors(default_nq, default_dim) # Phase 1: add property self.release_collection(client, col) self.drop_index(client, col, vec_field) self.alter_collection_properties(client, col, properties={"query_mode": "large_topk"}) index_params = self.prepare_index_params(client)[0] index_params.add_index(vec_field, index_type="FLAT", metric_type="L2") self.create_index(client, col, index_params) self.load_collection(client, col) self.search(client, col, data=vectors, anns_field=vec_field, limit=default_limit, check_task=CheckTasks.check_search_results, check_items={"enable_milvus_client_api": True, "nq": default_nq, "limit": default_limit, "metric": "L2"}) # Phase 2: drop property self.release_collection(client, col) self.drop_index(client, col, vec_field) self.drop_collection_properties(client, col, property_keys=["query_mode"]) self.create_index(client, col, index_params) self.load_collection(client, col) self.search(client, col, data=vectors, anns_field=vec_field, limit=default_limit, check_task=CheckTasks.check_search_results, check_items={"enable_milvus_client_api": True, "nq": default_nq, "limit": default_limit, "metric": "L2"}) desc = client.describe_collection(col) assert "query_mode" not in desc.get("properties", {}) @pytest.mark.tags(CaseLabel.L1) def test_property_persistence_after_reload(self): """ target: verify query_mode=large_topk persists after release + load method: 1. create collection with property, insert total > large_topk_first rows, flush 2. release → load 3. describe_collection to verify property present 4. search with limit=large_topk_first, verify large_topk_first results returned expected: property present; topk=large_topk_first returns exactly large_topk_first results """ client = self._client() nb_total = large_topk_first + 1000 col = self._setup_col(client, enable_large_topk=True, nb=nb_total) self.release_collection(client, col) self.load_collection(client, col) desc = client.describe_collection(col) assert desc.get("properties", {}).get("query_mode") == "large_topk", \ f"property lost after reload: {desc.get('properties')}" self.search(client, col, data=cf.gen_vectors(default_nq, default_dim), anns_field=vec_field, limit=large_topk_first, check_task=CheckTasks.check_search_results, check_items={"enable_milvus_client_api": True, "nq": default_nq, "limit": large_topk_first, "metric": "L2"}) @pytest.mark.tags(CaseLabel.L1) def test_create_index_after_insert_with_large_topk(self): """ target: verify create_index on a populated collection with query_mode=large_topk, and that large topk search (>16384) works correctly after index build method: 1. create collection with query_mode=large_topk (no index yet) 2. insert large_topk_first + default_nb rows, flush 3. create_index on the populated collection 4. load 5. search with limit=large_topk_first (>16384) to verify large topk is functional 6. search with limit > nb rows (capped to actual nb) to verify normal search works expected: create_index succeeds on populated collection; large topk search returns large_topk_first results; normal search returns min(limit, nb) results note: this catches the timeout seen when rebuilding index after alter/drop property, isolating whether the issue is data-at-index-build-time or the alter step itself """ client = self._client() nb = large_topk_first + default_nb col = cf.gen_collection_name_by_testcase_name() schema = self.create_schema(client)[0] schema.add_field("id", DataType.INT64, is_primary=True, auto_id=True) schema.add_field(vec_field, DataType.FLOAT_VECTOR, dim=default_dim) self.create_collection(client, col, schema=schema, properties={"query_mode": "large_topk"}, force_teardown=True) rows = [{vec_field: cf.gen_vectors(1, default_dim)[0]} for _ in range(nb)] self.insert(client, col, rows) self.flush(client, col) index_params = self.prepare_index_params(client)[0] index_params.add_index(vec_field, index_type="FLAT", metric_type="L2") self.create_index(client, col, index_params) self.load_collection(client, col) # verify large topk (>16384) is functional self.search(client, col, data=cf.gen_vectors(default_nq, default_dim), anns_field=vec_field, limit=large_topk_first, check_task=CheckTasks.check_search_results, check_items={"enable_milvus_client_api": True, "nq": default_nq, "limit": large_topk_first, "metric": "L2"}) # verify normal search also works self.search(client, col, data=cf.gen_vectors(default_nq, default_dim), anns_field=vec_field, limit=100, check_task=CheckTasks.check_search_results, check_items={"enable_milvus_client_api": True, "nq": default_nq, "limit": 100, "metric": "L2"}) @pytest.mark.tags(CaseLabel.L1) def test_large_topk_growing_segment(self): """ target: verify large_topk works on growing segments (before flush) method: 1. create collection with property, build FLAT index, load 2. insert default_nb rows WITHOUT flush (growing segment) 3. search with limit=100 — should return results from growing segment expected: search succeeds and returns hits; no error about topk limit """ client = self._client() col = self._setup_col(client, enable_large_topk=True, nb=0) # Insert without flush → growing segment rows = [{vec_field: cf.gen_vectors(1, default_dim)[0]} for _ in range(default_nb)] self.insert(client, col, rows) # No flush — data stays in growing segment self.search(client, col, data=cf.gen_vectors(default_nq, default_dim), anns_field=vec_field, limit=100, check_task=CheckTasks.check_search_results, check_items={"enable_milvus_client_api": True, "nq": default_nq, "limit": 100, "metric": "L2"}) @pytest.mark.tags(CaseLabel.L2) def test_invalid_property_value(self): """ target: verify invalid query_mode value is rejected at collection creation method: create_collection with properties={"query_mode": "invalid_mode"} expected: MilvusException with message containing valid values hint """ client = self._client() col = cf.gen_collection_name_by_testcase_name() schema = self.create_schema(client)[0] schema.add_field("id", DataType.INT64, is_primary=True, auto_id=True) schema.add_field(vec_field, DataType.FLOAT_VECTOR, dim=default_dim) error = {ct.err_code: 65535, ct.err_msg: 'invalid query_mode value "invalid_mode", valid values: [large_topk]'} self.create_collection(client, col, schema=schema, properties={"query_mode": "invalid_mode"}, check_task=CheckTasks.err_res, check_items=error) @pytest.mark.tags(CaseLabel.L2) @pytest.mark.parametrize("value", ["LARGE_TOPK", "Large_TopK", "large_TOPK"]) def test_query_mode_value_case_insensitive(self, value): """ target: verify query_mode value is case-sensitive method: create collection with properties={"query_mode": value} (non-lowercase value) expected: create collection fails with invalid query_mode value error """ client = self._client() col = cf.gen_collection_name_by_testcase_name() schema = self.create_schema(client)[0] schema.add_field("id", DataType.INT64, is_primary=True, auto_id=True) schema.add_field(vec_field, DataType.FLOAT_VECTOR, dim=default_dim) error = {ct.err_code: 65535, ct.err_msg: f'invalid query_mode value "{value}", valid values: [large_topk]'} self.create_collection(client, col, schema=schema, properties={"query_mode": value}, check_task=CheckTasks.err_res, check_items=error) @pytest.mark.tags(CaseLabel.L2) @pytest.mark.parametrize("key", ["QUERY_MODE", "Query_Mode", "query_MODE"]) def test_query_mode_key_case_sensitive(self, key): """ target: verify query_mode key is case-sensitive method: create collection with properties={key: "large_topk"} (wrong-cased key) expected: create collection fails with invalid property key error """ client = self._client() col = cf.gen_collection_name_by_testcase_name() schema = self.create_schema(client)[0] schema.add_field("id", DataType.INT64, is_primary=True, auto_id=True) schema.add_field(vec_field, DataType.FLOAT_VECTOR, dim=default_dim) error = {ct.err_code: 65535, ct.err_msg: f'invalid property key "{key}", did you mean "query_mode"?'} self.create_collection(client, col, schema=schema, properties={key: "large_topk"}, check_task=CheckTasks.err_res, check_items=error) # Note: search_iterator and query_iterator are NOT affected by query_mode=large_topk. # The SDK enforces batch_size <= 16384 client-side (ParamError code=1, regardless of property). # Iterator `limit` (total results) uses internal pagination with batch_size <= 16384 per request, # so per-request topk never exceeds 16384. No large_topk-specific iterator tests are needed. # ------------------------------------------------------------------------- # Large topk boundary tests (L3, large data volume) # ------------------------------------------------------------------------- @pytest.mark.tags(CaseLabel.L3) def test_large_topk_boundary_2m_rows(self): """ target: verify large_topk topk boundary values with 2M rows method: 1. create collection with query_mode=large_topk 2. insert 2,000,000 rows (128-dim) in batches, flush 3. create_index, load 4. search with limit=large_topk_max-1 (999,999) → should succeed 5. search with limit=large_topk_max (1,000,000) → should succeed, return 1M results 6. search with limit=large_topk_max+1 (1,000,001) → should fail with error expected: boundary limits enforced correctly; max valid topk returns 1M results """ client = self._client() total_nb = 2_000_000 batch_size = 50_000 col = cf.gen_collection_name_by_testcase_name() dim = 64 schema = self.create_schema(client)[0] schema.add_field("id", DataType.INT64, is_primary=True, auto_id=True) schema.add_field(vec_field, DataType.FLOAT_VECTOR, dim=dim) self.create_collection(client, col, schema=schema, properties={"query_mode": "large_topk"}, force_teardown=True) for _ in range(total_nb // batch_size): vecs = cf.gen_vectors(batch_size, dim) rows = [{vec_field: v} for v in vecs] self.insert(client, col, rows) self.flush(client, col) index_params = self.prepare_index_params(client)[0] index_params.add_index(vec_field, index_type="FLAT", metric_type="L2") self.create_index(client, col, index_params) self.load_collection(client, col) # just below max → should succeed self.search(client, col, data=cf.gen_vectors(1, dim), anns_field=vec_field, limit=large_topk_max - 1, check_task=CheckTasks.check_search_results, check_items={"enable_milvus_client_api": True, "nq": 1, "limit": large_topk_max - 1, "metric": "L2"}) # max valid large topk → should succeed, return 1M results self.search(client, col, data=cf.gen_vectors(1, dim), anns_field=vec_field, limit=large_topk_max, check_task=CheckTasks.check_search_results, check_items={"enable_milvus_client_api": True, "nq": 1, "limit": large_topk_max, "metric": "L2"}) # over max → should fail error = {ct.err_code: 65535, ct.err_msg: f"topk [{large_topk_max + 1}] is invalid, " f"it should be in range [1, {large_topk_max}]"} self.search(client, col, data=cf.gen_vectors(1, dim), anns_field=vec_field, limit=large_topk_max + 1, check_task=CheckTasks.err_res, check_items=error) # ------------------------------------------------------------------------- # Hybrid search interface tests # ------------------------------------------------------------------------- def _setup_dual_vec_col(self, client, enable_large_topk=True, nb=default_nb): """Create collection with two float vector fields for hybrid search tests. Uses IVF_FLAT index to keep index-build time reasonable for large nb.""" col = cf.gen_collection_name_by_testcase_name(module_index=2) schema = self.create_schema(client)[0] schema.add_field("id", DataType.INT64, is_primary=True, auto_id=True) schema.add_field(vec_field, DataType.FLOAT_VECTOR, dim=default_dim) schema.add_field("vec2", DataType.FLOAT_VECTOR, dim=default_dim) query_mode_props = {"query_mode": "large_topk"} if enable_large_topk else None self.create_collection(client, col, schema=schema, properties=query_mode_props, force_teardown=True) index_params = self.prepare_index_params(client)[0] index_params.add_index(vec_field, index_type="IVF_FLAT", metric_type="L2", params={"nlist": 64}) index_params.add_index("vec2", index_type="IVF_FLAT", metric_type="L2", params={"nlist": 64}) self.create_index(client, col, index_params) self.load_collection(client, col) if nb > 0: rows = [{vec_field: cf.gen_vectors(1, default_dim)[0], "vec2": cf.gen_vectors(1, default_dim)[0]} for _ in range(nb)] self.insert(client, col, rows) self.flush(client, col) return col @pytest.mark.tags(CaseLabel.L1) def test_hybrid_search_large_topk(self): """ target: verify hybrid_search with limit > 16384 works when query_mode=large_topk is set method: 1. create collection with two float vector fields and query_mode=large_topk 2. insert large_topk_total rows with IVF_FLAT index 3. hybrid_search with limit=large_topk_first using RRFRanker expected: hybrid_search completes without error; returns > 0 results; no error code """ client = self._client() col = self._setup_dual_vec_col(client, enable_large_topk=True, nb=large_topk_total) req_list = [ AnnSearchRequest( data=cf.gen_vectors(default_nq, default_dim), anns_field=vec_field, param={"metric_type": "L2", "nprobe": 16}, limit=large_topk_first, ), AnnSearchRequest( data=cf.gen_vectors(default_nq, default_dim), anns_field="vec2", param={"metric_type": "L2", "nprobe": 16}, limit=large_topk_first, ), ] res, _ = self.hybrid_search(client, col, reqs=req_list, ranker=RRFRanker(), limit=large_topk_first) assert len(res) == default_nq, f"Expected {default_nq} query results, got {len(res)}" for hits in res: assert len(hits) > 0, "Expected non-empty hybrid search results" @pytest.mark.tags(CaseLabel.L2) def test_hybrid_search_without_property_fails(self): """ target: verify hybrid_search with limit > 16384 is rejected when query_mode=large_topk is NOT set method: 1. create collection with two float vector fields and NO query_mode property 2. hybrid_search with limit=large_topk_first expected: MilvusException with invalid topk message """ client = self._client() col = self._setup_dual_vec_col(client, enable_large_topk=False) req_list = [ AnnSearchRequest( data=cf.gen_vectors(default_nq, default_dim), anns_field=vec_field, param={"metric_type": "L2"}, limit=large_topk_first, ), AnnSearchRequest( data=cf.gen_vectors(default_nq, default_dim), anns_field="vec2", param={"metric_type": "L2"}, limit=large_topk_first, ), ] # hybrid_search uses "invalid max query result window" (not "topk [N] is invalid") error = {ct.err_code: 65535, ct.err_msg: f"invalid max query result window, (offset+limit) should be in range [1, 16384], but got {large_topk_first}"} self.hybrid_search(client, col, reqs=req_list, ranker=RRFRanker(), limit=large_topk_first, check_task=CheckTasks.err_res, check_items=error)