test: add struct array null semantics and lifecycle coverage (#51524)

issue: #51381

related: #51414
related: #51416

### What this PR does

- Adds Python client E2E coverage for nullable Struct Array
NULL-versus-empty semantics.
- Covers compaction, snapshot restoration, partial updates, drop and
re-add isolation, imports, regex indexes, aggregation, and dynamic
fields.
- Adds exact ANN oracles for Struct Array element and embedding-list
searches without assuming self-hit behavior.
- Covers hybrid search collapse strategies and element identity
preservation.
- Marks known regressions with strict xfail markers linked to their
corresponding issues while keeping unaffected parameters active.

### Why

Struct Array rows can represent NULL, empty arrays, and non-empty
arrays. These states affect predicates, nested indexes, element offsets,
and vector search candidates differently. The added coverage protects
these semantics across write, reload, compaction, import, and snapshot
lifecycle paths.

### Validation

- `ruff check` passed for all changed Python files.
- `ruff format --check` passed for all changed Python files.
- Python compilation passed for all changed Python files.
- Pytest collection passed.
- The nine known regression nodes executed as expected: `9 xfailed`.
- The three unaffected parameterized cases passed.
- Targeted E2E validation used Milvus master build `09c44f2dbb`.

---------

Signed-off-by: zhuwenxing <wenxing.zhu@zilliz.com>

tests/python_client/milvus_client/test_drop_field_feature.py

@@ -1084,6 +1084,138 @@ class TestMilvusClientDropFieldFeature(TestMilvusClientV2Base):
         assert "events" not in new_row[0]
         self.drop_collection(client, collection_name)
 
+    @pytest.mark.tags(CaseLabel.L1)
+    def test_drop_then_readd_same_struct_array_name_isolates_old_offsets(self):
+        """
+        target: verify a same-name Struct Array added after drop gets new field IDs and independent physical data
+        method: drop an indexed vector Struct, re-add the same parent/child name with a scalar-only schema through
+            an alias, then query old/new rows before and after nested index build and reload
+        expected: old Struct binlogs and indexes never surface through the new field; old rows are null and new offsets
+            belong only to newly inserted rows
+        """
+        client = self._client()
+        collection_name = cf.gen_collection_name_by_testcase_name()
+        alias = cf.gen_unique_str("struct_readd_alias")
+        schema = client.create_schema(enable_dynamic_field=False, auto_id=False)
+        schema.add_field("id", DataType.INT64, is_primary=True)
+        schema.add_field("vec", DataType.FLOAT_VECTOR, dim=4)
+        old_struct_schema = client.create_struct_field_schema()
+        old_struct_schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=4)
+        old_struct_schema.add_field("name", DataType.VARCHAR, max_length=64)
+        schema.add_field(
+            "events",
+            datatype=DataType.ARRAY,
+            element_type=DataType.STRUCT,
+            struct_schema=old_struct_schema,
+            max_capacity=4,
+        )
+        index_params = client.prepare_index_params()
+        index_params.add_index("vec", index_type="HNSW", metric_type="L2", params={"M": 8, "efConstruction": 64})
+        index_params.add_index("events[embedding]", index_type="HNSW", metric_type="MAX_SIM_L2")
+        index_params.add_index("events[name]", index_type="INVERTED")
+        client.create_collection(
+            collection_name,
+            schema=schema,
+            index_params=index_params,
+            consistency_level="Strong",
+        )
+        client.create_alias(collection_name, alias)
+
+        old_rows = [
+            {
+                "id": row_id,
+                "vec": [float(row_id), 0.0, 0.0, 0.0],
+                "events": [
+                    {
+                        "embedding": [float(row_id), 1.0, 0.0, 0.0],
+                        "name": f"old_{row_id}",
+                    }
+                ],
+            }
+            for row_id in range(3)
+        ]
+        client.insert(collection_name, old_rows)
+        client.flush(collection_name)
+        client.load_collection(collection_name)
+        before_drop = client.describe_collection(collection_name)
+        old_parent = next(field for field in before_drop["fields"] if field["name"] == "events")
+        old_parent_id = old_parent.get("field_id")
+
+        client.drop_collection_field(collection_name, "events")
+        for _ in range(30):
+            if not any(index_name.startswith("events[") for index_name in client.list_indexes(collection_name)):
+                break
+            time.sleep(1)
+        assert not any(index_name.startswith("events[") for index_name in client.list_indexes(collection_name))
+
+        new_struct_schema = client.create_struct_field_schema()
+        new_struct_schema.add_field("name", DataType.VARCHAR, max_length=128)
+        new_struct_schema.add_field("rank", DataType.INT64)
+        client.add_collection_struct_field(
+            alias,
+            "events",
+            new_struct_schema,
+            max_capacity=3,
+        )
+        after_add = client.describe_collection(alias)
+        new_parent = next(field for field in after_add["fields"] if field["name"] == "events")
+        new_parent_id = new_parent.get("field_id")
+        if old_parent_id is not None and new_parent_id is not None:
+            assert new_parent_id > old_parent_id
+
+        old_after_add = client.query(
+            alias,
+            filter="id in [0, 1, 2]",
+            output_fields=["id", "events"],
+        )
+        assert {row["id"] for row in old_after_add} == {0, 1, 2}
+        assert all(row["events"] is None for row in old_after_add)
+        assert client.query(alias, filter='array_contains(events[name], "old_1")', output_fields=["id"]) == []
+
+        new_rows = [
+            {"id": 100, "vec": [100.0, 0.0, 0.0, 0.0], "events": []},
+            {
+                "id": 101,
+                "vec": [101.0, 0.0, 0.0, 0.0],
+                "events": [{"name": "new_a", "rank": 10}, {"name": "new_b", "rank": 20}],
+            },
+        ]
+        client.insert(alias, new_rows)
+        client.flush(alias)
+        nested_indexes = client.prepare_index_params()
+        nested_indexes.add_index("events[name]", index_type="BITMAP")
+        nested_indexes.add_index("events[rank]", index_type="STL_SORT")
+        client.create_index(alias, nested_indexes)
+
+        def assert_new_schema_data():
+            old_rows_result = client.query(
+                alias,
+                filter="id in [0, 1, 2]",
+                output_fields=["id", "events"],
+            )
+            assert all(row["events"] is None for row in old_rows_result)
+            contains = client.query(
+                alias,
+                filter='array_contains(events[name], "new_b")',
+                output_fields=["id", "events"],
+            )
+            assert [row["id"] for row in contains] == [101]
+            assert contains[0]["events"] == new_rows[1]["events"]
+            element_rows = client.query(
+                alias,
+                filter="element_filter(events, $[rank] >= 10)",
+                output_fields=["id"],
+                limit=10,
+            )
+            assert sorted((row["id"], row["offset"]) for row in element_rows) == [(101, 0), (101, 1)]
+
+        assert_new_schema_data()
+        client.release_collection(alias)
+        client.load_collection(alias)
+        assert_new_schema_data()
+        client.drop_alias(alias)
+        self.drop_collection(client, collection_name)
+
     @pytest.mark.tags(CaseLabel.L1)
     def test_drop_field_negative_constraint_matrix(self):
         """

tests/python_client/milvus_client/test_milvus_client_partial_update.py

@@ -8,6 +8,7 @@ from common import common_type as ct
 from common.common_type import CaseLabel, CheckTasks
 from pymilvus import DataType, Function, FunctionType
 from pymilvus.client.embedding_list import EmbeddingList
+from pymilvus.client.field_ops import FieldOp
 from sklearn import preprocessing
 from utils.util_log import test_log as log
 from utils.util_pymilvus import *  # noqa: F403
@@ -369,6 +370,210 @@ class TestMilvusClientPartialUpdateValid(TestMilvusClientV2Base):
             assert search_results[0][0]["payload"] == f"payload_{row_id}"
             self._assert_struct_array_partial_update_events_equal(search_results[0][0]["events"], expected_events)
 
+    @pytest.mark.tags(CaseLabel.L1)
+    def test_milvus_client_partial_update_struct_array_state_transitions(self):
+        """
+        target: verify whole-Struct partial updates preserve null, empty, and element offset semantics
+        method: transition sealed rows through null, empty, and different non-empty lengths while omitting
+            ordinary fields, then compare nested-index query and element search before/after compact and reload
+        expected: the whole Struct is replaced, omitted ordinary fields are preserved, and stale elements disappear
+        """
+        client = self._client()
+        dim = 8
+        collection_name = cf.gen_unique_str(f"{prefix}_pu_struct_states")
+        schema = self.create_schema(client, enable_dynamic_field=False)[0]
+        schema.add_field("id", DataType.INT64, is_primary=True, auto_id=False)
+        schema.add_field("vector", DataType.FLOAT_VECTOR, dim=dim)
+        schema.add_field("payload", DataType.VARCHAR, max_length=64)
+        struct_schema = self.create_struct_field_schema(client)[0]
+        struct_schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=dim)
+        struct_schema.add_field("score", DataType.INT64)
+        struct_schema.add_field("tag", DataType.VARCHAR, max_length=64)
+        schema.add_field(
+            "events",
+            datatype=DataType.ARRAY,
+            element_type=DataType.STRUCT,
+            struct_schema=struct_schema,
+            max_capacity=4,
+            nullable=True,
+        )
+
+        index_params = self.prepare_index_params(client)[0]
+        index_params.add_index("vector", index_type="HNSW", metric_type="COSINE", params={"M": 8, "efConstruction": 64})
+        index_params.add_index(
+            "events[embedding]", index_type="HNSW", metric_type="COSINE", params={"M": 8, "efConstruction": 64}
+        )
+        index_params.add_index("events[tag]", index_type="BITMAP")
+        self.create_collection(
+            client,
+            collection_name,
+            schema=schema,
+            index_params=index_params,
+            consistency_level="Strong",
+        )
+
+        def vector(axis):
+            value = [0.0] * dim
+            value[axis % dim] = 1.0
+            return value
+
+        def events(row_id, count, state):
+            return [
+                {
+                    "embedding": vector(row_id + offset),
+                    "score": row_id * 100 + offset,
+                    "tag": f"{state}_{row_id}_{offset}",
+                }
+                for offset in range(count)
+            ]
+
+        expected = {
+            0: {"id": 0, "vector": vector(0), "payload": "payload_0", "events": None},
+            1: {"id": 1, "vector": vector(1), "payload": "payload_1", "events": []},
+            2: {"id": 2, "vector": vector(2), "payload": "payload_2", "events": events(2, 1, "initial")},
+            3: {"id": 3, "vector": vector(3), "payload": "payload_3", "events": events(3, 2, "initial")},
+        }
+        self.insert(client, collection_name, list(expected.values()))
+        self.flush(client, collection_name)
+        self.load_collection(client, collection_name)
+
+        first_transition = {
+            0: [],
+            1: events(1, 1, "first"),
+            2: None,
+            3: events(3, 3, "first"),
+        }
+        self.upsert(
+            client,
+            collection_name,
+            [{"id": row_id, "events": value} for row_id, value in first_transition.items()],
+            partial_update=True,
+        )
+        for row_id, value in first_transition.items():
+            expected[row_id]["events"] = value
+        self.flush(client, collection_name)
+
+        second_transition = {
+            0: events(0, 2, "final"),
+            1: None,
+            2: [],
+            3: events(3, 1, "final"),
+        }
+        self.upsert(
+            client,
+            collection_name,
+            [{"id": row_id, "events": value} for row_id, value in second_transition.items()],
+            partial_update=True,
+        )
+        for row_id, value in second_transition.items():
+            expected[row_id]["events"] = value
+        self.flush(client, collection_name)
+
+        def collect_observations():
+            rows = self.query(
+                client,
+                collection_name,
+                filter="id >= 0",
+                output_fields=["id", "vector", "payload", "events"],
+                limit=len(expected),
+            )[0]
+            rows_by_id = {row["id"]: row for row in rows}
+            assert set(rows_by_id) == set(expected)
+            for row_id, expected_row in expected.items():
+                assert rows_by_id[row_id]["payload"] == expected_row["payload"]
+                assert np.allclose(rows_by_id[row_id]["vector"], expected_row["vector"])
+                self._assert_struct_array_partial_update_events_equal(
+                    rows_by_id[row_id]["events"] or [], expected_row["events"] or []
+                )
+                assert (rows_by_id[row_id]["events"] is None) == (expected_row["events"] is None)
+
+            indexed_rows = self.query(
+                client,
+                collection_name,
+                filter='array_contains(events[tag], "final_0_1")',
+                output_fields=["id"],
+                limit=len(expected),
+            )[0]
+            element_hits = self.search(
+                client,
+                collection_name,
+                data=[vector(0)],
+                anns_field="events[embedding]",
+                search_params={"metric_type": "COSINE", "params": {"ef": 64}},
+                filter="element_filter(events, $[score] >= 0)",
+                output_fields=["id"],
+                limit=3,
+            )[0][0]
+            return {
+                "projection": sorted(
+                    (row_id, rows_by_id[row_id]["payload"], rows_by_id[row_id]["events"]) for row_id in rows_by_id
+                ),
+                "indexed_ids": sorted(row["id"] for row in indexed_rows),
+                "element_keys": sorted((hit["id"], hit["offset"]) for hit in element_hits),
+            }
+
+        baseline = collect_observations()
+        assert baseline["indexed_ids"] == [0]
+        assert baseline["element_keys"] == [(0, 0), (0, 1), (3, 0)]
+
+        compact_id = self.compact(client, collection_name)[0]
+        assert compact_id > 0
+        assert self.wait_for_compaction_ready(client, compact_id, timeout=300)
+        self.release_collection(client, collection_name)
+        self.load_collection(client, collection_name)
+        assert collect_observations() == baseline
+
+    @pytest.mark.tags(CaseLabel.L2)
+    @pytest.mark.parametrize(
+        "field_name",
+        ["events", "events[tag]"],
+        ids=["struct-parent", "struct-subfield"],
+    )
+    def test_milvus_client_partial_update_struct_array_field_ops_rejected(self, field_name):
+        """
+        target: verify Array partial-update operators cannot mutate Struct parents or Struct sub-fields
+        method: apply ARRAY_APPEND to a whole Struct field and to one child field
+        expected: both requests are rejected and the original Struct remains unchanged
+        """
+        client = self._client()
+        collection_name = cf.gen_unique_str(f"{prefix}_pu_struct_op")
+        schema = self.create_schema(client, enable_dynamic_field=False)[0]
+        schema.add_field("id", DataType.INT64, is_primary=True, auto_id=False)
+        schema.add_field("vector", DataType.FLOAT_VECTOR, dim=4)
+        struct_schema = self.create_struct_field_schema(client)[0]
+        struct_schema.add_field("tag", DataType.VARCHAR, max_length=32)
+        schema.add_field(
+            "events",
+            datatype=DataType.ARRAY,
+            element_type=DataType.STRUCT,
+            struct_schema=struct_schema,
+            max_capacity=4,
+            nullable=True,
+        )
+        index_params = self.prepare_index_params(client)[0]
+        index_params.add_index("vector", index_type="HNSW", metric_type="COSINE", params={"M": 8, "efConstruction": 64})
+        self.create_collection(
+            client, collection_name, schema=schema, index_params=index_params, consistency_level="Strong"
+        )
+        original = {"id": 0, "vector": [1.0, 0.0, 0.0, 0.0], "events": [{"tag": "original"}]}
+        self.insert(client, collection_name, [original])
+        self.load_collection(client, collection_name)
+
+        error = {
+            ct.err_code: 1100,
+            ct.err_msg: f'op ARRAY_APPEND is not supported for struct field "{field_name}"',
+        }
+        self.upsert(
+            client,
+            collection_name,
+            [{"id": 0, "events": [{"tag": "new"}]}],
+            field_ops={field_name: FieldOp.array_append()},
+            check_task=CheckTasks.err_res,
+            check_items=error,
+        )
+        rows = self.query(client, collection_name, filter="id == 0", output_fields=["events"])[0]
+        assert rows[0]["events"] == original["events"]
+
     @pytest.mark.tags(CaseLabel.L0)
     def test_partial_update_all_field_types_one_by_one(self):
         """

tests/python_client/milvus_client/test_milvus_client_regex_filter.py

@@ -2455,6 +2455,57 @@ class TestRegexFilterStructArray(RegexFilterStructArraySharedBase):
         result = sorted([r["id"] for r in res])
         assert result == [1], f"indexed struct name =~ error.*timeout: expected [1], got {result}"
 
+    @pytest.mark.tags(CaseLabel.L1)
+    def test_regex_struct_array_raw_index_reload_consistency(self):
+        """
+        target: regex results remain stable across raw, nested scalar-indexed, and reloaded states
+        expected: MATCH and element_filter PK/offset results are identical before indexing, after indexing, and reload
+        """
+        client = self._client()
+        collection_name = cf.gen_collection_name_by_testcase_name()
+        setup_struct_array_regex_collection(client, collection_name, create_scalar_index=False)
+
+        def collect_observations():
+            match_name = client.query(
+                collection_name,
+                filter='MATCH_ANY(events, $[name] =~ "error.*timeout")',
+                output_fields=["id"],
+            )
+            match_status = client.query(
+                collection_name,
+                filter='MATCH_ANY(events, $[status] !~ "ERROR")',
+                output_fields=["id"],
+            )
+            matching_elements = client.query(
+                collection_name,
+                filter='element_filter(events, $[name] =~ "login.*")',
+                output_fields=["id"],
+                limit=100,
+            )
+            return {
+                "match_name": sorted(row["id"] for row in match_name),
+                "match_status": sorted(row["id"] for row in match_status),
+                "elements": sorted((row["id"], row["offset"]) for row in matching_elements),
+            }
+
+        baseline = collect_observations()
+        assert baseline == {
+            "match_name": [1],
+            "match_status": [1, 2, 3],
+            "elements": [(1, 0), (1, 1)],
+        }
+
+        index_params = client.prepare_index_params()
+        index_params.add_index(field_name="events[name]", index_type="INVERTED")
+        index_params.add_index(field_name="events[status]", index_type="INVERTED")
+        client.create_index(collection_name, index_params)
+        assert collect_observations() == baseline
+
+        client.release_collection(collection_name)
+        client.load_collection(collection_name)
+        assert collect_observations() == baseline
+        self.drop_collection(client, collection_name)
+
     @pytest.mark.tags(CaseLabel.L1)
     def test_regex_struct_array_hybrid_search(self):
         """

tests/python_client/milvus_client/test_milvus_client_search_aggregation.py

@@ -1617,9 +1617,9 @@ class TestSearchAggregationIndependent(TestMilvusClientV2Base):
     @pytest.mark.tags(CaseLabel.L2)
     def test_struct_array_element_search_with_primary_key_aggregation(self):
         """
-        target: clarify StructArray element-vector search support when grouped by primary key.
-        method: search structA[embedding] and request aggregation by the primary key field.
-        expected: request succeeds because element-level StructArray search only supports primary-key grouping.
+        target: verify StructArray element results are aggregated exactly by primary key.
+        method: search six elements from three primary keys and request PK buckets with counts and two top hits.
+        expected: all three PK buckets contain exactly their two elements without cross-row mixing or loss.
         """
         client = self._client()
         collection_name = cf.gen_collection_name_by_testcase_name()
@@ -1646,16 +1646,24 @@ class TestSearchAggregationIndependent(TestMilvusClientV2Base):
             params={"M": 8, "efConstruction": 64},
         )
         self.create_collection(client, collection_name, schema=schema, index_params=index_params)
-        vectors = cf.gen_vectors(self.nb * 2, dim=dim, vector_data_type=DataType.FLOAT_VECTOR)
+        vectors = [
+            [1.0, 0.0] + [0.0] * (dim - 2),
+            [0.9, 0.1] + [0.0] * (dim - 2),
+            [0.0, 1.0] + [0.0] * (dim - 2),
+            [0.1, 0.9] + [0.0] * (dim - 2),
+            [-1.0, 0.0] + [0.0] * (dim - 2),
+            [-0.9, -0.1] + [0.0] * (dim - 2),
+        ]
+        primary_keys = [10, 20, 30]
         rows = [
             {
-                "id": i,
+                "id": primary_key,
                 "structA": [
                     {"embedding": vectors[i * 2], "label": f"label_{i}_0"},
                     {"embedding": vectors[i * 2 + 1], "label": f"label_{i}_1"},
                 ],
             }
-            for i in range(self.nb)
+            for i, primary_key in enumerate(primary_keys)
         ]
         self.insert(client, collection_name, data=rows)
         self.flush(client, collection_name)
@@ -1664,25 +1672,30 @@ class TestSearchAggregationIndependent(TestMilvusClientV2Base):
         res, _ = self.search(
             client,
             collection_name,
-            data=[vectors[0]],
+            data=[[0.7, 0.7] + [0.0] * (dim - 2)],
             anns_field="structA[embedding]",
-            search_params={"metric_type": "COSINE", "params": {"ef": 32}},
-            limit=10,
-            output_fields=["id"],
+            search_params={"metric_type": "COSINE", "params": {"ef": 64}},
+            limit=len(vectors),
+            output_fields=["id", "structA[label]"],
             search_aggregation=SearchAggregation(
                 fields=["id"],
                 size=3,
                 metrics={"doc_count": {"count": "*"}},
-                top_hits=TopHits(size=1),
+                top_hits=TopHits(size=2, sort=[{"_score": "desc"}]),
             ),
         )
 
         assert len(res.agg_buckets) == 1
-        assert 1 <= len(res.agg_buckets[0]) <= 3
+        assert len(res.agg_buckets[0]) == len(primary_keys)
+        buckets_by_id = {bucket.key[0]["value"]: bucket for bucket in res.agg_buckets[0]}
+        assert set(buckets_by_id) == set(primary_keys)
         for bucket in res.agg_buckets[0]:
             assert [entry["field_name"] for entry in bucket.key] == ["id"]
-            assert bucket.metrics["doc_count"] == bucket.count
-            assert len(bucket.hits) == 1
+            primary_key = bucket.key[0]["value"]
+            assert bucket.metrics["doc_count"] == bucket.count == 2
+            assert len(bucket.hits) == 2
+            assert all(hit.pk == primary_key for hit in bucket.hits)
+            assert bucket.hits[0].score >= bucket.hits[1].score
 
 
 @pytest.mark.xdist_group("TestSearchAggregationTextAndBM25")

tests/python_client/milvus_client/test_milvus_client_snapshot.py

@@ -9,6 +9,7 @@ from common import common_type as ct
 from common.common_type import CaseLabel, CheckTasks
 from ml_dtypes import bfloat16
 from pymilvus import DataType
+from pymilvus.client.embedding_list import EmbeddingList
 from utils.util_log import test_log as log
 
 prefix = "snapshot"
@@ -2243,6 +2244,152 @@ class TestMilvusClientSnapshotAllDataTypes(TestMilvusClientSnapshotBase):
         self.drop_snapshot(client, snapshot_name, collection_name)
         self.drop_collection(client, restored_collection_name)
 
+    @pytest.mark.tags(CaseLabel.L1)
+    def test_snapshot_restores_struct_array_semantics_and_indexes(self):
+        """
+        target: verify snapshot restores Struct Array schema, nullability, nested indexes, and both search modes
+        method: snapshot nullable Struct rows with empty and variable-length values plus separate embedding-list and
+            element-vector sub-fields, restore to a new collection, and compare exact semantic observations
+        expected: restored projection, predicates, complete valid ANN candidate sets, and index metadata equal source
+        """
+        client = self._client()
+        collection_name = cf.gen_unique_str(f"{prefix}_struct_source")
+        restored_collection_name = cf.gen_unique_str(f"{prefix}_struct_restored")
+        snapshot_name = cf.gen_unique_str(f"{prefix}_struct")
+        dim = 8
+
+        schema = client.create_schema(auto_id=False, enable_dynamic_field=False)
+        schema.add_field("id", DataType.INT64, is_primary=True)
+        schema.add_field("vector", DataType.FLOAT_VECTOR, dim=dim)
+        struct_schema = client.create_struct_field_schema()
+        struct_schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=dim)
+        struct_schema.add_field("element_embedding", DataType.FLOAT_VECTOR, dim=dim)
+        struct_schema.add_field("tag", DataType.VARCHAR, max_length=32)
+        schema.add_field(
+            "events",
+            datatype=DataType.ARRAY,
+            element_type=DataType.STRUCT,
+            struct_schema=struct_schema,
+            max_capacity=4,
+            nullable=True,
+        )
+        index_params = client.prepare_index_params()
+        index_params.add_index("vector", index_type="HNSW", metric_type="COSINE", params={"M": 8, "efConstruction": 64})
+        index_params.add_index(
+            "events[embedding]",
+            index_type="HNSW",
+            metric_type="MAX_SIM_COSINE",
+            params={"M": 8, "efConstruction": 64},
+        )
+        index_params.add_index(
+            "events[element_embedding]",
+            index_type="HNSW",
+            metric_type="COSINE",
+            params={"M": 8, "efConstruction": 64},
+        )
+        index_params.add_index("events[tag]", index_type="BITMAP")
+        self.create_collection(
+            client,
+            collection_name,
+            schema=schema,
+            index_params=index_params,
+            consistency_level="Strong",
+        )
+
+        def vector(axis):
+            value = [0.0] * dim
+            value[axis % dim] = 1.0
+            return value
+
+        def event(axis, tag):
+            value = vector(axis)
+            return {"embedding": value, "element_embedding": value, "tag": tag}
+
+        rows = [
+            {"id": 0, "vector": vector(0), "events": None},
+            {"id": 1, "vector": vector(1), "events": []},
+            {"id": 2, "vector": vector(2), "events": [event(0, "target"), event(2, "other")]},
+            {"id": 3, "vector": vector(3), "events": [event(1, "control")]},
+        ]
+        self.insert(client, collection_name, rows)
+        self.flush(client, collection_name)
+        self.load_collection(client, collection_name)
+
+        def collect_observations(target_collection):
+            projected = self.query(
+                client,
+                target_collection,
+                filter="id >= 0",
+                output_fields=["id", "events"],
+                limit=len(rows),
+            )[0]
+            contains = self.query(
+                client,
+                target_collection,
+                filter='array_contains(events[tag], "target")',
+                output_fields=["id"],
+                limit=len(rows),
+            )[0]
+            matched = self.query(
+                client,
+                target_collection,
+                filter='MATCH_ANY(events, $[tag] == "target")',
+                output_fields=["id"],
+                limit=len(rows),
+            )[0]
+            element_hits = self.search(
+                client,
+                target_collection,
+                data=[vector(0)],
+                anns_field="events[element_embedding]",
+                search_params={"metric_type": "COSINE", "params": {"ef": 64}},
+                output_fields=["id"],
+                limit=3,
+            )[0][0]
+            tensor = EmbeddingList()
+            tensor.add(vector(0))
+            embedding_hits = self.search(
+                client,
+                target_collection,
+                data=[tensor],
+                anns_field="events[embedding]",
+                search_params={"metric_type": "MAX_SIM_COSINE", "params": {"ef": 32}},
+                output_fields=["id"],
+                limit=2,
+            )[0][0]
+            return {
+                "projection": sorted((row["id"], row["events"]) for row in projected),
+                "contains_ids": sorted(row["id"] for row in contains),
+                "match_ids": sorted(row["id"] for row in matched),
+                "element_keys": sorted((hit["id"], hit["offset"]) for hit in element_hits),
+                "embedding_ids": sorted(hit["id"] for hit in embedding_hits),
+                "indexes": sorted(client.list_indexes(target_collection)),
+            }
+
+        source_observations = collect_observations(collection_name)
+        assert source_observations["contains_ids"] == [2]
+        assert source_observations["match_ids"] == [2]
+        assert source_observations["element_keys"] == [(2, 0), (2, 1), (3, 0)]
+        assert source_observations["embedding_ids"] == [2, 3]
+
+        self.create_snapshot(client, snapshot_name, collection_name)
+        job_id = self.restore_snapshot(
+            client,
+            snapshot_name,
+            collection_name,
+            restored_collection_name,
+        )[0]
+        wait_for_restore_complete(self, client, job_id)
+        self.load_collection(client, restored_collection_name)
+        try:
+            restored_observations = collect_observations(restored_collection_name)
+            for key in ("projection", "contains_ids", "match_ids", "element_keys", "indexes"):
+                assert restored_observations[key] == source_observations[key]
+            assert restored_observations["embedding_ids"] == [2, 3]
+        finally:
+            self.drop_snapshot(client, snapshot_name, collection_name)
+            self.drop_collection(client, restored_collection_name)
+
     @pytest.mark.tags(CaseLabel.L2)
     def test_snapshot_with_bfloat16_vector(self):
         """

tests/python_client/milvus_client/test_milvus_client_struct_array_element_search.py

@@ -732,6 +732,72 @@ class TestMilvusClientStructArrayElementFilterSearch(TestMilvusClientV2Base):
         assert "offset" in top_hit, "element offset not exposed on hit"
         assert top_hit["offset"] == target_elem, f"expected element offset={target_elem}, got {top_hit['offset']}"
 
+    @pytest.mark.tags(CaseLabel.L1)
+    def test_element_filter_template_matches_inline_query_and_search(self):
+        """
+        target: verify filter templates bind Struct child values for element-level query and search
+        method: compare a string template against its inline element_filter form for query and vector search
+        expected: template and inline forms return identical PK/offset results
+        """
+        client = self._client()
+        collection_name = cf.gen_unique_str(f"{prefix}_ef_template")
+        data = self._create_collection_and_insert(client, collection_name, nb=20, metric_type="COSINE")
+        target_row = 5
+        target_elem = 2
+        tag = data[target_row]["structA"][target_elem]["str_val"]
+        query_vector = data[target_row]["structA"][target_elem]["embedding"]
+        inline = f'element_filter(structA, $[str_val] == "{tag}")'
+        template = "element_filter(structA, $[str_val] == {tag})"
+        failures = []
+
+        inline_query = client.query(
+            collection_name,
+            filter=inline,
+            output_fields=["id"],
+            limit=100,
+        )
+        try:
+            template_query = client.query(
+                collection_name,
+                filter=template,
+                filter_params={"tag": tag},
+                output_fields=["id"],
+                limit=100,
+            )
+            assert [(row["id"], row["offset"]) for row in template_query] == [
+                (row["id"], row["offset"]) for row in inline_query
+            ]
+        except Exception as exc:
+            failures.append(f"query failed: {exc}")
+
+        inline_search = client.search(
+            collection_name,
+            data=[query_vector],
+            anns_field="structA[embedding]",
+            search_params={"metric_type": "COSINE"},
+            filter=inline,
+            limit=10,
+            output_fields=["id"],
+        )[0]
+        try:
+            template_search = client.search(
+                collection_name,
+                data=[query_vector],
+                anns_field="structA[embedding]",
+                search_params={"metric_type": "COSINE"},
+                filter=template,
+                filter_params={"tag": tag},
+                limit=10,
+                output_fields=["id"],
+            )[0]
+            assert [(hit["id"], hit["offset"]) for hit in template_search] == [
+                (hit["id"], hit["offset"]) for hit in inline_search
+            ]
+        except Exception as exc:
+            failures.append(f"search failed: {exc}")
+
+        assert not failures, "\n".join(failures)
+
     # ---- L1 tests ----
 
     @pytest.mark.tags(CaseLabel.L1)
@@ -3788,7 +3854,12 @@ class TestMilvusClientStructArrayElementHybridSearch(TestMilvusClientV2Base):
 
         index_params = client.prepare_index_params()
         for field_name in ["normal_vector", "structA[embedding]", "structA[embedding_alt]", "structB[embedding]"]:
-            index_params.add_index(field_name=field_name, index_type="FLAT", metric_type=metric_type)
+            index_params.add_index(
+                field_name=field_name,
+                index_type="HNSW",
+                metric_type=metric_type,
+                params={"M": 8, "efConstruction": 64},
+            )
         self.create_collection(client, collection_name, schema=schema, index_params=index_params)
 
         def _struct_a(scores):
@@ -4007,9 +4078,9 @@ class TestMilvusClientStructArrayElementHybridSearch(TestMilvusClientV2Base):
         query_vector = self._cosine_vector(1.0)
 
         def _req(field_name, params=None):
-            search_params = {"metric_type": "COSINE"}
+            search_params = {"metric_type": "COSINE", "params": {"ef": 64}}
             if params is not None:
-                search_params["params"] = params
+                search_params["params"].update(params)
             return AnnSearchRequest(
                 **{
                     "data": [query_vector],
@@ -4046,6 +4117,237 @@ class TestMilvusClientStructArrayElementHybridSearch(TestMilvusClientV2Base):
         assert scoped_ids != default_ids
         assert len(scoped_ids) == len(set(scoped_ids)), f"row-level hybrid should not duplicate ids: {scoped_ids}"
 
+    @pytest.mark.tags(CaseLabel.L1)
+    def test_hybrid_element_scope_all_collapse_strategies_exact(self):
+        """
+        target: verify all Struct element collapse strategies with exact row order and RRF score
+        method: use identical deterministic vectors in two parent Struct fields, run nq=2 hybrid search for
+            max, sum, avg, topk_sum, and topk_avg, including topk larger than the row element count
+        expected: each strategy produces its calculated row order, exact RRF scores, and no element offset leakage
+        """
+        client = self._client()
+        collection_name = cf.gen_unique_str(f"{prefix}_hyb_scope_all")
+        schema = client.create_schema(auto_id=False, enable_dynamic_field=False)
+        schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
+
+        for struct_name in ("structA", "structB"):
+            struct_schema = client.create_struct_field_schema()
+            struct_schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=default_dim)
+            schema.add_field(
+                struct_name,
+                datatype=DataType.ARRAY,
+                element_type=DataType.STRUCT,
+                struct_schema=struct_schema,
+                max_capacity=4,
+            )
+
+        index_params = client.prepare_index_params()
+        for field_name in ("structA[embedding]", "structB[embedding]"):
+            index_params.add_index(
+                field_name=field_name,
+                index_type="HNSW",
+                metric_type="COSINE",
+                params={"M": 8, "efConstruction": 64},
+            )
+        self.create_collection(client, collection_name, schema=schema, index_params=index_params)
+
+        scores_by_id = {
+            1: [0.99, 0.10, 0.10, 0.10],
+            2: [0.80, 0.80, 0.80, 0.80],
+            3: [0.90, 0.90],
+            4: [0.95],
+        }
+        rows = []
+        for row_id, scores in scores_by_id.items():
+            values = [{"embedding": self._cosine_vector(score)} for score in scores]
+            rows.append({"id": row_id, "structA": values, "structB": values})
+        self.insert(client, collection_name, rows)
+        self.flush(client, collection_name)
+        self.load_collection(client, collection_name)
+
+        expected_orders = {
+            "max": [1, 4, 3, 2],
+            "sum": [2, 3, 1, 4],
+            "avg": [4, 3, 2, 1],
+            "topk_sum": [3, 2, 1, 4],
+            "topk_avg": [4, 3, 2, 1],
+            "topk_sum_all": [2, 3, 1, 4],
+            "topk_avg_all": [4, 3, 2, 1],
+        }
+        cases = [
+            ("max", {"strategy": "max"}),
+            ("sum", {"strategy": "sum"}),
+            ("avg", {"strategy": "avg"}),
+            ("topk_sum", {"strategy": "topk_sum", "topk": 2}),
+            ("topk_avg", {"strategy": "topk_avg", "topk": 2}),
+            ("topk_sum_all", {"strategy": "topk_sum", "topk": 10}),
+            ("topk_avg_all", {"strategy": "topk_avg", "topk": 10}),
+        ]
+        query_vector = self._cosine_vector(1.0)
+        rrf_k = 60
+
+        for case_name, collapse in cases:
+            requests = [
+                AnnSearchRequest(
+                    **{
+                        "data": [query_vector, query_vector],
+                        "anns_field": field_name,
+                        "param": {
+                            "metric_type": "COSINE",
+                            "params": {"ef": 64, "element_scope": {"collapse": collapse}},
+                        },
+                        "limit": sum(len(values) for values in scores_by_id.values()),
+                    }
+                )
+                for field_name in ("structA[embedding]", "structB[embedding]")
+            ]
+            results, check = self.hybrid_search(
+                client,
+                collection_name,
+                requests,
+                ranker=RRFRanker(rrf_k),
+                limit=len(scores_by_id),
+                output_fields=["id"],
+            )
+            assert check
+            assert len(results) == 2
+            for hits in results:
+                assert [hit["id"] for hit in hits] == expected_orders[case_name]
+                assert all("offset" not in hit for hit in hits)
+                for rank, hit in enumerate(hits):
+                    assert hit["distance"] == pytest.approx(2 / (rrf_k + rank + 1), abs=1e-6)
+
+    @pytest.mark.tags(CaseLabel.L2)
+    def test_hybrid_element_scope_validation_matrix(self):
+        """
+        target: verify element collapse strategy and topk validation
+        method: submit missing, zero, extraneous, and unsupported collapse parameters
+        expected: every invalid configuration is rejected with its source-defined parameter error
+        """
+        client = self._client()
+        collection_name = cf.gen_unique_str(f"{prefix}_hyb_scope_validation")
+        self._setup_multi_struct_collection(client, collection_name)
+        query_vector = self._cosine_vector(1.0)
+        invalid_cases = [
+            ({"strategy": "topk_sum"}, "element_scope.collapse.topk is required for strategy topk_sum"),
+            ({"strategy": "topk_avg", "topk": 0}, "element_scope.collapse.topk must be positive"),
+            ({"strategy": "max", "topk": 1}, "element_scope.collapse.topk is only valid for topk strategies"),
+            ({"strategy": "median"}, "unsupported element_scope.collapse.strategy: median"),
+        ]
+
+        for collapse, expected_message in invalid_cases:
+            req1 = AnnSearchRequest(
+                **{
+                    "data": [query_vector],
+                    "anns_field": "structA[embedding]",
+                    "param": {
+                        "metric_type": "COSINE",
+                        "params": {"ef": 64, "element_scope": {"collapse": collapse}},
+                    },
+                    "limit": 9,
+                }
+            )
+            req2 = AnnSearchRequest(
+                **{
+                    "data": [query_vector],
+                    "anns_field": "structB[embedding]",
+                    "param": {"metric_type": "COSINE", "params": {"ef": 64}},
+                    "limit": 9,
+                }
+            )
+            self.hybrid_search(
+                client,
+                collection_name,
+                [req1, req2],
+                ranker=RRFRanker(),
+                limit=3,
+                check_task=CheckTasks.err_res,
+                check_items={ct.err_code: 1100, ct.err_msg: expected_message},
+            )
+
+    @pytest.mark.tags(CaseLabel.L1)
+    @pytest.mark.parametrize(
+        "ranker",
+        [RRFRanker(60), WeightedRanker(0.5, 0.5)],
+        ids=["rrf", "weighted"],
+    )
+    def test_hybrid_same_struct_string_pk_element_identity_exact(self, ranker):
+        """
+        target: verify same-Struct hybrid preserves string primary key and element offset identity
+        method: rerank two identical vector sub-fields with RRF and WeightedRanker
+        expected: both rankers return the exact ordered (string PK, offset) sequence without row collapse
+        """
+        client = self._client()
+        collection_name = cf.gen_unique_str(f"{prefix}_hyb_string_pk")
+        schema = client.create_schema(auto_id=False, enable_dynamic_field=False)
+        schema.add_field(field_name="id", datatype=DataType.VARCHAR, max_length=32, is_primary=True)
+        struct_schema = client.create_struct_field_schema()
+        struct_schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=default_dim)
+        struct_schema.add_field("embedding_alt", DataType.FLOAT_VECTOR, dim=default_dim)
+        schema.add_field(
+            "structA",
+            datatype=DataType.ARRAY,
+            element_type=DataType.STRUCT,
+            struct_schema=struct_schema,
+            max_capacity=4,
+        )
+        index_params = client.prepare_index_params()
+        for field_name in ("structA[embedding]", "structA[embedding_alt]"):
+            index_params.add_index(
+                field_name=field_name,
+                index_type="HNSW",
+                metric_type="COSINE",
+                params={"M": 8, "efConstruction": 64},
+            )
+        self.create_collection(client, collection_name, schema=schema, index_params=index_params)
+
+        score_rows = {"doc/a": [0.99, 0.10], "doc:b": [0.90, 0.80]}
+        rows = []
+        for row_id, scores in score_rows.items():
+            rows.append(
+                {
+                    "id": row_id,
+                    "structA": [
+                        {
+                            "embedding": self._cosine_vector(score),
+                            "embedding_alt": self._cosine_vector(score),
+                        }
+                        for score in scores
+                    ],
+                }
+            )
+        self.insert(client, collection_name, rows)
+        self.flush(client, collection_name)
+        self.load_collection(client, collection_name)
+
+        query_vector = self._cosine_vector(1.0)
+        requests = [
+            AnnSearchRequest(
+                **{
+                    "data": [query_vector],
+                    "anns_field": field_name,
+                    "param": {"metric_type": "COSINE", "params": {"ef": 64}},
+                    "limit": 4,
+                }
+            )
+            for field_name in ("structA[embedding]", "structA[embedding_alt]")
+        ]
+        results, check = self.hybrid_search(
+            client,
+            collection_name,
+            requests,
+            ranker=ranker,
+            limit=4,
+            output_fields=["id"],
+        )
+        assert check
+        assert [(hit["id"], hit["offset"]) for hit in results[0]] == [
+            ("doc/a", 0),
+            ("doc:b", 0),
+            ("doc:b", 1),
+            ("doc/a", 1),
+        ]
+
     @pytest.mark.tags(CaseLabel.L1)
     def test_hybrid_element_level_range_same_struct_supported(self):
         """
@@ -4057,7 +4359,7 @@ class TestMilvusClientStructArrayElementHybridSearch(TestMilvusClientV2Base):
         collection_name = cf.gen_unique_str(f"{prefix}_hyb_range_same_struct")
         self._setup_multi_struct_collection(client, collection_name)
         query_vector = self._cosine_vector(1.0)
-        range_params = {"radius": 0.85}
+        range_params = {"ef": 64, "radius": 0.85}
 
         req1 = AnnSearchRequest(
             **{