"""
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)

