import time 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 from common.external_table_common import ( build_external_source, build_external_spec, get_minio_config, ) from pymilvus import DataType, Function, FunctionType def rows_by_id(rows): return {row["id"]: row for row in rows} class TestMilvusClientDropFieldFeature(TestMilvusClientV2Base): @pytest.mark.tags(CaseLabel.L0) def test_drop_scalar_field_basic_read_write(self): """ TC-L0-01: Drop normal scalar field and verify basic read/write paths. target: verify basic scalar drop field path method: drop VarChar field tag, then query/search/insert without it expected: tag disappears from schema; non-tag read/write works; tag references fail """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Step 1: Create a collection with the scalar field that will be dropped. 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) schema.add_field("age", DataType.INT64) schema.add_field("tag", DataType.VARCHAR, max_length=64) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) # Step 2: Insert sealed data containing the target field. rows = [{"id": i, "vec": [float(i), 0.0, 0.0, 0.0], "age": 20 + i, "tag": f"tag_{i}"} for i in range(5)] client.insert(collection_name=collection_name, data=rows) client.flush(collection_name) client.load_collection(collection_name) # Step 3: Drop the scalar field and verify it leaves the schema. client.drop_collection_field(collection_name, "tag") schema_fields = [field["name"] for field in client.describe_collection(collection_name)["fields"]] assert "tag" not in schema_fields assert "age" in schema_fields assert "vec" in schema_fields # Step 4: Verify read paths that do not reference the dropped field still work. query_res = client.query( collection_name, filter="id in [0, 1, 2, 3, 4]", output_fields=["id", "age"], ) query_by_id = rows_by_id(query_res) assert query_by_id[0]["age"] == 20 assert query_by_id[4]["age"] == 24 search_res = client.search( collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=3, output_fields=["id", "age"], ) assert len(search_res[0]) == 3 assert "age" in search_res[0][0]["entity"] assert "tag" not in search_res[0][0]["entity"] # Step 5: Verify explicit references to the dropped field fail clearly. for kwargs in [ {"filter": 'tag == "tag_1"', "output_fields": ["id"]}, {"filter": "id >= 0", "output_fields": ["id", "tag"]}, ]: self.query( client, collection_name, check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "field tag not exist"}, **kwargs, ) # Step 6: Insert new rows without the dropped field. client.insert( collection_name=collection_name, data=[{"id": 100, "vec": [1.0, 0.0, 0.0, 0.0], "age": 100}], ) client.flush(collection_name) new_row = client.query(collection_name, filter="id == 100", output_fields=["*"]) assert rows_by_id(new_row)[100]["age"] == 100 assert "tag" not in rows_by_id(new_row)[100] self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L0) def test_drop_indexed_scalar_field_cascade(self): """ TC-L0-02: Drop indexed scalar field and verify index cascade convergence. target: verify index metadata on dropped field is removed method: build indexes on vec and tag, drop tag, then verify tag index disappears expected: tag field and tag index disappear; vec index and vec search keep working """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Step 1: Create a collection with indexes on both the kept vector field and the dropped scalar field. 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) schema.add_field("age", DataType.INT64) schema.add_field("tag", DataType.VARCHAR, max_length=64) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") index_params.add_index(field_name="tag", index_type="INVERTED") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) # Step 2: Insert enough sealed rows for the scalar index to be materialized. rows = [ { "id": i, "vec": [float(i % 10), float(i % 7), float(i % 5), float(i % 3)], "age": 20 + (i % 100), "tag": f"tag_{i % 50}", } for i in range(3000) ] client.insert(collection_name=collection_name, data=rows) client.flush(collection_name) assert self.wait_for_index_ready(client, collection_name, index_name="vec", timeout=120) assert self.wait_for_index_ready(client, collection_name, index_name="tag", timeout=120) indexes = client.list_indexes(collection_name) assert "vec" in indexes assert "tag" in indexes tag_index = client.describe_index(collection_name, index_name="tag") assert tag_index["field_name"] == "tag" assert tag_index["index_type"] == "INVERTED" # Step 3: Confirm the indexed field is usable before drop. client.load_collection(collection_name) pre_drop_query = client.query( collection_name, filter='tag == "tag_1"', output_fields=["id", "tag"], limit=5, ) assert len(pre_drop_query) > 0 assert all(row["tag"] == "tag_1" for row in pre_drop_query) # Step 4: Drop the indexed scalar field. client.drop_collection_field(collection_name, "tag") schema_fields = [field["name"] for field in client.describe_collection(collection_name)["fields"]] assert "tag" not in schema_fields assert "vec" in schema_fields assert "age" in schema_fields # Step 5: Wait for index metadata cascade and verify the vector index remains. for _ in range(30): indexes = client.list_indexes(collection_name) if "tag" not in indexes: break time.sleep(1) assert "tag" not in indexes assert "vec" in indexes vec_index = client.describe_index(collection_name, index_name="vec") assert vec_index["field_name"] == "vec" # Step 6: Verify read paths on the kept fields still work. search_res = client.search( collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=5, output_fields=["id", "age"], ) assert len(search_res[0]) == 5 assert "age" in search_res[0][0]["entity"] assert "tag" not in search_res[0][0]["entity"] # Step 7: Verify dropped field references and dropped index lookup fail. self.query( client, collection_name, filter='tag == "tag_1"', output_fields=["id"], check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "field tag not exist"}, ) assert client.describe_index(collection_name, index_name="tag") is None self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L0) def test_drop_bm25_function_removes_output_field_and_index(self): """ TC-L0-03: Drop BM25 function and verify output field/index cascade. target: verify dropping BM25 function removes function output field and its index method: create BM25 function from text to sparse, reject detach-only drop, then cascade-drop function field expected: function and sparse output field disappear; text and vec remain usable """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Step 1: Create a BM25 function whose output sparse vector field will be cascade-dropped. schema = client.create_schema(enable_dynamic_field=False, auto_id=False) schema.add_field("id", DataType.INT64, is_primary=True) schema.add_field("text", DataType.VARCHAR, max_length=1024, enable_analyzer=True) schema.add_field("sparse", DataType.SPARSE_FLOAT_VECTOR) schema.add_field("vec", DataType.FLOAT_VECTOR, dim=4) schema.add_function( Function( name="bm25", function_type=FunctionType.BM25, input_field_names=["text"], output_field_names="sparse", ) ) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") index_params.add_index(field_name="sparse", index_type="AUTOINDEX", metric_type="BM25") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) # Step 2: Insert text rows and wait for both dense and sparse indexes. rows = [ { "id": i, "text": f"milvus bm25 function document {i % 5}", "vec": [float(i % 10), float(i % 7), float(i % 5), float(i % 3)], } for i in range(100) ] client.insert(collection_name=collection_name, data=rows) client.flush(collection_name) assert self.wait_for_index_ready(client, collection_name, index_name="vec", timeout=120) assert self.wait_for_index_ready(client, collection_name, index_name="sparse", timeout=120) indexes = client.list_indexes(collection_name) assert "vec" in indexes assert "sparse" in indexes desc = client.describe_collection(collection_name) assert [func["name"] for func in desc.get("functions", [])] == ["bm25"] assert "sparse" in [field["name"] for field in desc["fields"]] # Step 3: Confirm BM25 search works before dropping the function. client.load_collection(collection_name) bm25_res = client.search( collection_name, data=["milvus function"], anns_field="sparse", limit=5, output_fields=["id", "text"], ) assert len(bm25_res[0]) > 0 assert "text" in bm25_res[0][0]["entity"] # Step 4: BM25 cannot be detached from its output field through the legacy SDK API. self.drop_collection_function( client, collection_name, "bm25", check_task=ct.CheckTasks.err_res, check_items={ ct.err_code: 1100, ct.err_msg: ( "BM25 function must be dropped with its output field in drop_function_field interface: " "bm25: invalid parameter" ), }, ) desc = client.describe_collection(collection_name) field_names = [field["name"] for field in desc["fields"]] function_names = [func["name"] for func in desc.get("functions", [])] assert "bm25" in function_names assert "sparse" in field_names assert "sparse" in client.list_indexes(collection_name) # Step 5: Drop through the cascade SDK API and verify its output field and index disappear. self.drop_function_field(client, collection_name, "bm25") desc = client.describe_collection(collection_name) field_names = [field["name"] for field in desc["fields"]] function_names = [func["name"] for func in desc.get("functions", [])] assert "bm25" not in function_names assert "sparse" not in field_names assert "text" in field_names assert "vec" in field_names indexes = client.list_indexes(collection_name) assert "sparse" not in indexes assert "vec" in indexes assert client.describe_index(collection_name, index_name="sparse") is None # Step 6: Verify remaining scalar and dense vector read paths still work. query_res = client.query( collection_name, filter="id in [0, 1, 2]", output_fields=["id", "text"], ) assert len(query_res) == 3 assert all("text" in row for row in query_res) vec_res = client.search( collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=5, output_fields=["id", "text"], ) assert len(vec_res[0]) == 5 assert "text" in vec_res[0][0]["entity"] assert "sparse" not in vec_res[0][0]["entity"] # Step 7: Verify the dropped sparse field cannot be searched. self.search( client, collection_name, data=["milvus function"], anns_field="sparse", limit=5, output_fields=["id"], check_task=ct.CheckTasks.err_res, check_items={ ct.err_code: 1100, ct.err_msg: "failed to get field schema by name: fieldName(sparse) not found", }, ) # Step 8: Insert rows using the remaining fields. client.insert( collection_name=collection_name, data=[{"id": 1000, "text": "text after bm25 drop", "vec": [1.0, 0.0, 0.0, 0.0]}], ) new_row = client.query(collection_name, filter="id == 1000", output_fields=["*"]) assert new_row[0]["text"] == "text after bm25 drop" assert "sparse" not in new_row[0] self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L0) def test_disable_dynamic_schema_removes_dynamic_visibility(self): """ TC-L0-04: Disable dynamic schema and verify dynamic data is no longer visible. target: verify disabling dynamic schema hides old dynamic keys and rejects new unknown keys method: insert dynamic rows, disable dynamic schema, then query/search/insert around dynamic keys expected: dynamic keys disappear from API output; dynamic references fail; static fields still work """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Step 1: Create a dynamic-schema collection with static fields only in the declared schema. schema = client.create_schema(enable_dynamic_field=True, auto_id=False) schema.add_field("id", DataType.INT64, is_primary=True) schema.add_field("vec", DataType.FLOAT_VECTOR, dim=4) schema.add_field("age", DataType.INT64) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) # Step 2: Insert rows with dynamic keys and verify they are visible before disable. rows = [ { "id": i, "vec": [float(i), 0.0, 0.0, 0.0], "age": 20 + i, "dyn_tag": f"tag_{i}", "dyn_score": i, } for i in range(5) ] client.insert(collection_name=collection_name, data=rows) client.flush(collection_name) client.load_collection(collection_name) before_desc = client.describe_collection(collection_name) assert before_desc["enable_dynamic_field"] is True before_meta = client.query( collection_name, filter="id == 1", output_fields=["id", "$meta"], ) assert before_meta[0]["dyn_tag"] == "tag_1" assert before_meta[0]["dyn_score"] == 1 before_query = client.query( collection_name, filter='dyn_tag == "tag_1"', output_fields=["id", "dyn_tag"], ) assert len(before_query) == 1 assert before_query[0]["dyn_tag"] == "tag_1" before_star = client.query(collection_name, filter="id == 1", output_fields=["*"]) assert before_star[0]["dyn_tag"] == "tag_1" assert before_star[0]["dyn_score"] == 1 # Step 3: Disable dynamic schema and verify dynamic metadata is removed from schema output. client.alter_collection_properties(collection_name, {"dynamicfield.enabled": False}) after_desc = client.describe_collection(collection_name) assert after_desc["enable_dynamic_field"] is False assert "$meta" not in [field["name"] for field in after_desc["fields"]] # Step 4: Verify wildcard query/search no longer exposes old dynamic keys. after_star = client.query( collection_name, filter="id in [0, 1, 2, 3, 4]", output_fields=["*"], ) assert len(after_star) == 5 for row in after_star: assert "id" in row assert "age" in row assert "vec" in row assert "dyn_tag" not in row assert "dyn_score" not in row assert "$meta" not in row search_res = client.search( collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=3, output_fields=["*"], ) assert len(search_res[0]) == 3 assert "age" in search_res[0][0]["entity"] assert "dyn_tag" not in search_res[0][0]["entity"] assert "dyn_score" not in search_res[0][0]["entity"] assert "$meta" not in search_res[0][0]["entity"] # Step 5: Verify explicit dynamic key and $meta references fail. for kwargs in [ {"filter": 'dyn_tag == "tag_1"', "output_fields": ["id"]}, {"filter": "id >= 0", "output_fields": ["id", "dyn_tag"]}, {"filter": "id >= 0", "output_fields": ["id", "$meta"]}, ]: expected_field = "$meta" if "$meta" in kwargs["output_fields"] else "dyn_tag" self.query( client, collection_name, check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: f"field {expected_field} not exist"}, **kwargs, ) # Step 6: Verify static-only insert works and new dynamic keys are rejected. client.insert( collection_name=collection_name, data=[{"id": 100, "vec": [1.0, 0.0, 0.0, 0.0], "age": 100}], ) static_row = client.query(collection_name, filter="id == 100", output_fields=["*"]) assert static_row[0]["age"] == 100 assert "dyn_after_disable" not in static_row[0] self.insert( client, collection_name=collection_name, data=[ { "id": 101, "vec": [1.0, 1.0, 0.0, 0.0], "age": 101, "dyn_after_disable": "new", } ], check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1, ct.err_msg: "Attempt to insert an unexpected field `dyn_after_disable`"}, ) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_drop_one_vector_field_keeps_another_vector_field(self): """ TC-L1-01: Drop one vector field and keep another vector field. target: verify multi-vector collection keeps remaining vector field searchable after dropping one vector field method: create dense1 and dense2, build indexes, drop dense2, then search dense1 and reject dense2/last-vector drop expected: dense2 disappears; dense1 search works; dense2 search fails; dropping dense1 is rejected as last vector field """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Step 1: Create a collection with two vector fields. schema = client.create_schema(enable_dynamic_field=False, auto_id=False) schema.add_field("id", DataType.INT64, is_primary=True) schema.add_field("dense1", DataType.FLOAT_VECTOR, dim=4) schema.add_field("dense2", DataType.FLOAT_VECTOR, dim=4) schema.add_field("age", DataType.INT64) index_params = client.prepare_index_params() index_params.add_index(field_name="dense1", index_type="AUTOINDEX", metric_type="L2") index_params.add_index(field_name="dense2", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) # Step 2: Insert sealed rows and wait for both vector indexes. rows = [ { "id": i, "dense1": [float(i % 10), float(i % 7), float(i % 5), float(i % 3)], "dense2": [float(i % 3), float(i % 5), float(i % 7), float(i % 10)], "age": 20 + (i % 100), } for i in range(3000) ] client.insert(collection_name=collection_name, data=rows) client.flush(collection_name) assert self.wait_for_index_ready(client, collection_name, index_name="dense1", timeout=120) assert self.wait_for_index_ready(client, collection_name, index_name="dense2", timeout=120) indexes = client.list_indexes(collection_name) assert "dense1" in indexes assert "dense2" in indexes client.load_collection(collection_name) # Step 3: Confirm the vector field that will be dropped is searchable before drop. dense2_before_drop = client.search( collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="dense2", limit=5, output_fields=["id", "age"], ) assert len(dense2_before_drop[0]) == 5 assert "age" in dense2_before_drop[0][0]["entity"] # Step 4: Drop one vector field and verify only that field and index disappear. client.drop_collection_field(collection_name, "dense2") desc = client.describe_collection(collection_name) field_names = [field["name"] for field in desc["fields"]] assert "dense1" in field_names assert "dense2" not in field_names assert "age" in field_names for _ in range(30): indexes = client.list_indexes(collection_name) if "dense2" not in indexes: break time.sleep(1) assert "dense1" in indexes assert "dense2" not in indexes assert client.describe_index(collection_name, index_name="dense2") is None # Step 5: Verify the remaining vector field still supports search. dense1_res = client.search( collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="dense1", limit=5, output_fields=["id", "age"], ) assert len(dense1_res[0]) == 5 assert "age" in dense1_res[0][0]["entity"] assert "dense2" not in dense1_res[0][0]["entity"] # Step 6: Verify dropped vector search fails and the last vector field cannot be dropped. self.search( client, collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="dense2", limit=5, output_fields=["id"], check_task=ct.CheckTasks.err_res, check_items={ ct.err_code: 1100, ct.err_msg: "failed to get field schema by name: fieldName(dense2) not found", }, ) self.drop_collection_field( client, collection_name, field_name="dense1", check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "cannot drop the last vector field: dense1"}, ) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_drop_then_add_same_name_field_does_not_reuse_old_data(self): """ TC-L1-02: Drop then add same-name field and verify old data does not pollute the new field. target: verify same-name field after drop is a new schema field method: drop VarChar extra, add Int64 extra with same name, then query before and after reload expected: old extra values are not exposed through the new extra field """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Step 1: Create a collection with the original VarChar extra field. 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) schema.add_field("age", DataType.INT64) schema.add_field("extra", DataType.VARCHAR, max_length=64) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) # Step 2: Insert old sealed rows with a recognizable extra value. old_rows = [{"id": i, "vec": [float(i), 0.0, 0.0, 0.0], "age": 20 + i, "extra": "old_value"} for i in range(5)] client.insert(collection_name=collection_name, data=old_rows) client.flush(collection_name) client.load_collection(collection_name) before_drop_desc = client.describe_collection(collection_name) old_extra_field = next(field for field in before_drop_desc["fields"] if field["name"] == "extra") old_extra_field_id = old_extra_field.get("field_id") old_query = client.query(collection_name, filter='extra == "old_value"', output_fields=["id", "extra"]) assert len(old_query) == 5 assert all(row["extra"] == "old_value" for row in old_query) # Step 3: Drop extra and add a same-name field with a different type. client.drop_collection_field(collection_name, "extra") after_drop_desc = client.describe_collection(collection_name) assert "extra" not in [field["name"] for field in after_drop_desc["fields"]] max_field_id_after_drop = None properties = after_drop_desc.get("properties", {}) if isinstance(properties, dict) and properties.get("max_field_id") is not None: max_field_id_after_drop = int(properties["max_field_id"]) elif isinstance(properties, list): for prop in properties: if prop.get("key") == "max_field_id": max_field_id_after_drop = int(prop["value"]) break client.add_collection_field( collection_name, field_name="extra", data_type=DataType.INT64, nullable=True, ) after_add_desc = client.describe_collection(collection_name) new_extra_field = next(field for field in after_add_desc["fields"] if field["name"] == "extra") new_extra_field_id = new_extra_field.get("field_id") if old_extra_field_id is not None and new_extra_field_id is not None: assert new_extra_field_id > old_extra_field_id if max_field_id_after_drop is not None and new_extra_field_id is not None: assert new_extra_field_id > max_field_id_after_drop # Step 4: Verify old rows do not expose old physical extra data through the new extra field. old_rows_after_add = client.query( collection_name, filter="id in [0, 1, 2, 3, 4]", output_fields=["id", "extra"], ) assert len(old_rows_after_add) == 5 for row in old_rows_after_add: assert row.get("extra") != "old_value" # Step 5: Insert new rows using the new Int64 extra field. new_rows = [ {"id": 100 + i, "vec": [float(i), 1.0, 0.0, 0.0], "age": 100 + i, "extra": 2026 + i} for i in range(5) ] client.insert(collection_name=collection_name, data=new_rows) client.flush(collection_name) new_extra_query = client.query( collection_name, filter="extra >= 2026", output_fields=["id", "extra"], ) assert len(new_extra_query) == 5 assert {row["id"] for row in new_extra_query} == {100, 101, 102, 103, 104} assert {row["extra"] for row in new_extra_query} == {2026, 2027, 2028, 2029, 2030} # Step 6: Verify search output does not return old extra values. search_res = client.search( collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=10, output_fields=["id", "extra"], ) assert len(search_res[0]) == 10 for hit in search_res[0]: assert hit["entity"].get("extra") != "old_value" # Step 7: Reload and repeat the key old/new row checks. client.release_collection(collection_name) client.load_collection(collection_name) old_rows_after_reload = client.query( collection_name, filter="id in [0, 1, 2, 3, 4]", output_fields=["id", "extra"], ) assert len(old_rows_after_reload) == 5 for row in old_rows_after_reload: assert row.get("extra") != "old_value" new_extra_after_reload = client.query( collection_name, filter="extra >= 2026", output_fields=["id", "extra"], ) assert len(new_extra_after_reload) == 5 assert {row["id"] for row in new_extra_after_reload} == {100, 101, 102, 103, 104} self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_drop_field_loaded_collection_reload_delta_load_path(self): """ TC-L1-03: Loaded collection reload / delta-load path. target: verify dropped field is skipped across loaded sealed data, DDL-sealed growing data, and reload method: insert sealed rows, load, insert growing rows, drop tag, insert post-drop rows, then verify before and after reload expected: all rows remain visible through kept fields; tag is not visible or usable after drop """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Step 1: Create a collection with a field that will be dropped while the collection is loaded. 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) schema.add_field("age", DataType.INT64) schema.add_field("tag", DataType.VARCHAR, max_length=64) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) # Step 2: Insert and flush pre-drop sealed rows. sealed_rows = [ {"id": i, "vec": [float(i), 0.0, 0.0, 0.0], "age": 20 + i, "tag": "sealed_before_drop"} for i in range(5) ] client.insert(collection_name=collection_name, data=sealed_rows) client.flush(collection_name) client.load_collection(collection_name) sealed_check = client.query( collection_name, filter='tag == "sealed_before_drop"', output_fields=["id", "tag"], ) assert len(sealed_check) == 5 # Step 3: Insert pre-drop growing rows without manual flush. growing_rows = [ {"id": 100 + i, "vec": [float(i), 1.0, 0.0, 0.0], "age": 100 + i, "tag": "growing_before_drop"} for i in range(5) ] client.insert(collection_name=collection_name, data=growing_rows) growing_check = client.query( collection_name, filter='tag == "growing_before_drop"', output_fields=["id", "tag"], ) assert len(growing_check) == 5 # Step 4: Drop tag while sealed and growing pre-drop rows both exist. client.drop_collection_field(collection_name, "tag") desc = client.describe_collection(collection_name) assert "tag" not in [field["name"] for field in desc["fields"]] # Step 5: Verify pre-drop sealed and DDL-sealed growing rows remain visible through kept fields. pre_drop_rows = client.query( collection_name, filter="id in [0, 1, 2, 3, 4, 100, 101, 102, 103, 104]", output_fields=["id", "age"], ) assert {row["id"] for row in pre_drop_rows} == {0, 1, 2, 3, 4, 100, 101, 102, 103, 104} assert all("tag" not in row for row in pre_drop_rows) search_res = client.search( collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=10, output_fields=["id", "age"], ) assert len(search_res[0]) == 10 assert all("tag" not in hit["entity"] for hit in search_res[0]) # Step 6: Verify explicit references to the dropped field fail after drop. for kwargs in [ {"filter": 'tag == "sealed_before_drop"', "output_fields": ["id"]}, {"filter": "id >= 0", "output_fields": ["id", "tag"]}, ]: self.query( client, collection_name, check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "field tag not exist"}, **kwargs, ) # Step 7: Insert post-drop rows using the latest schema. post_drop_rows = [{"id": 200 + i, "vec": [float(i), 2.0, 0.0, 0.0], "age": 200 + i} for i in range(5)] client.insert(collection_name=collection_name, data=post_drop_rows) client.flush(collection_name) all_rows = client.query( collection_name, filter="id in [0, 1, 2, 3, 4, 100, 101, 102, 103, 104, 200, 201, 202, 203, 204]", output_fields=["id", "age"], ) assert {row["id"] for row in all_rows} == { 0, 1, 2, 3, 4, 100, 101, 102, 103, 104, 200, 201, 202, 203, 204, } assert all("tag" not in row for row in all_rows) # Step 8: Release and load again to verify segment reload skips dropped field data. client.release_collection(collection_name) client.load_collection(collection_name) rows_after_reload = client.query( collection_name, filter="id in [0, 1, 2, 3, 4, 100, 101, 102, 103, 104, 200, 201, 202, 203, 204]", output_fields=["id", "age"], ) assert {row["id"] for row in rows_after_reload} == { 0, 1, 2, 3, 4, 100, 101, 102, 103, 104, 200, 201, 202, 203, 204, } assert all("tag" not in row for row in rows_after_reload) search_after_reload = client.search( collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=10, output_fields=["id", "age"], ) assert len(search_after_reload[0]) == 10 assert all("tag" not in hit["entity"] for hit in search_after_reload[0]) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_drop_struct_array_field_and_reject_sub_field_drop(self): """ TC-L1-05: Drop StructArray field and reject sub-field drop. target: verify whole StructArray field can be dropped and StructArray sub-field drop is rejected method: reject events[name] drop first, then drop events and verify nested index cascade expected: failed sub-field drop does not change schema; events and nested indexes disappear after whole-field drop """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Step 1: Create a collection with a StructArray field and nested indexes. 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) struct_schema = client.create_struct_field_schema() struct_schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=4) struct_schema.add_field("name", DataType.VARCHAR, max_length=128) struct_schema.add_field("score", DataType.INT64) schema.add_field( "events", datatype=DataType.ARRAY, element_type=DataType.STRUCT, struct_schema=struct_schema, max_capacity=4, ) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") index_params.add_index(field_name="events[embedding]", index_type="HNSW", metric_type="MAX_SIM_L2") index_params.add_index(field_name="events[name]", index_type="INVERTED") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) # Step 2: Insert rows and verify the StructArray field is usable before any drop. rows = [ { "id": i, "vec": [float(i), 0.0, 0.0, 0.0], "events": [ { "embedding": [float(i), 0.0, 0.0, 0.0], "name": f"event_{i}", "score": i, }, { "embedding": [float(i), 1.0, 0.0, 0.0], "name": f"event_extra_{i}", "score": i + 100, }, ], } for i in range(10) ] client.insert(collection_name=collection_name, data=rows) client.flush(collection_name) assert self.wait_for_index_ready(client, collection_name, index_name="vec", timeout=60) indexes = client.list_indexes(collection_name) assert "vec" in indexes assert "events[embedding]" in indexes assert "events[name]" in indexes assert ( client.describe_index(collection_name, index_name="events[embedding]")["field_name"] == "events[embedding]" ) assert client.describe_index(collection_name, index_name="events[name]")["field_name"] == "events[name]" client.load_collection(collection_name) pre_drop_rows = client.query( collection_name, filter='array_contains(events[name], "event_1")', output_fields=["id", "events"], ) assert len(pre_drop_rows) == 1 assert pre_drop_rows[0]["events"][0]["name"] == "event_1" # Step 3: Reject direct sub-field drop and verify schema/indexes are unchanged. self.drop_collection_field( client, collection_name, field_name="events[name]", check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "field not found: events[name]: invalid parameter"}, ) desc = client.describe_collection(collection_name) field_names = [field["name"] for field in desc["fields"]] assert "events" in field_names indexes = client.list_indexes(collection_name) assert "events[embedding]" in indexes assert "events[name]" in indexes assert "vec" in indexes # Step 4: Drop the whole StructArray field. client.drop_collection_field(collection_name, "events") desc = client.describe_collection(collection_name) field_names = [field["name"] for field in desc["fields"]] assert "events" not in field_names assert "vec" in field_names # Step 5: Verify nested indexes are cascade-removed while the normal vector index remains. for _ in range(30): indexes = client.list_indexes(collection_name) if "events[embedding]" not in indexes and "events[name]" not in indexes: break time.sleep(1) assert "events[embedding]" not in indexes assert "events[name]" not in indexes assert "vec" in indexes assert client.describe_index(collection_name, index_name="events[embedding]") is None assert client.describe_index(collection_name, index_name="events[name]") is None # Step 6: Verify kept-field read/write paths still work and StructArray references fail. search_res = client.search( collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=5, output_fields=["id"], ) assert len(search_res[0]) == 5 self.query( client, collection_name, filter='array_contains(events[name], "event_1")', output_fields=["id"], check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "struct field not found: events[name]"}, ) client.insert( collection_name=collection_name, data=[{"id": 100, "vec": [1.0, 0.0, 0.0, 0.0]}], ) new_row = client.query(collection_name, filter="id == 100", output_fields=["*"]) assert new_row[0]["id"] == 100 assert "events" not in new_row[0] self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_drop_field_negative_constraint_matrix(self): """ TC-L1-06: Drop Field negative constraint matrix. target: verify Proxy rejects invalid drop-field targets before schema mutation method: attempt to drop protected, missing, system, last-vector, and function-referenced fields expected: each invalid request fails with the exact validation message and schema remains unchanged """ client = self._client() collection_name = f"{cf.gen_collection_name_by_testcase_name()}_matrix" last_vector_collection_name = f"{cf.gen_collection_name_by_testcase_name()}_last_vector" # Step 1: Create a collection that contains most protected field categories. schema = client.create_schema(enable_dynamic_field=True, auto_id=False) schema.add_field("id", DataType.INT64, is_primary=True) schema.add_field("vec", DataType.FLOAT_VECTOR, dim=4) schema.add_field("partition_tag", DataType.VARCHAR, max_length=64, is_partition_key=True) schema.add_field("cluster_score", DataType.INT64, is_clustering_key=True) schema.add_field("text", DataType.VARCHAR, max_length=1024, enable_analyzer=True) schema.add_field("sparse", DataType.SPARSE_FLOAT_VECTOR) schema.add_field("age", DataType.INT64) schema.add_function( Function( name="bm25", function_type=FunctionType.BM25, input_field_names=["text"], output_field_names="sparse", ) ) client.create_collection( collection_name=collection_name, schema=schema, consistency_level="Strong", ) # Step 2: Verify each invalid field target returns its exact validation reason. for field_name, expected_msg in [ ("", "Must specify exactly one valid Drop identifier (drop_field_name/drop_field_id/drop_function_name)"), ("missing_field", "field not found: missing_field"), ("id", "cannot drop primary key field: id"), ("partition_tag", "cannot drop partition key field: partition_tag"), ("cluster_score", "cannot drop clustering key field: cluster_score"), ("$rowid", "field not found: $rowid"), ("$timestamp", "field not found: $timestamp"), ("$namespace", "field not found: $namespace"), ("$meta", "field not found: $meta"), ("text", "field is referenced by function bm25 as input, drop function first"), ("sparse", "field is referenced by function bm25 as output, drop function first"), ]: self.drop_collection_field( client, collection_name, field_name=field_name, check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: expected_msg}, ) # Step 3: Verify the rejected attempts did not mutate the schema. desc = client.describe_collection(collection_name) field_names = [field["name"] for field in desc["fields"]] function_names = [func["name"] for func in desc.get("functions", [])] assert "id" in field_names assert "vec" in field_names assert "partition_tag" in field_names assert "cluster_score" in field_names assert "text" in field_names assert "sparse" in field_names assert "bm25" in function_names # Step 4: Create a separate collection for the last-vector constraint. last_vector_schema = client.create_schema(enable_dynamic_field=False, auto_id=False) last_vector_schema.add_field("id", DataType.INT64, is_primary=True) last_vector_schema.add_field("vec", DataType.FLOAT_VECTOR, dim=4) last_vector_schema.add_field("age", DataType.INT64) client.create_collection( collection_name=last_vector_collection_name, schema=last_vector_schema, consistency_level="Strong", ) self.drop_collection_field( client, last_vector_collection_name, field_name="vec", check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "cannot drop the last vector field: vec"}, ) # Step 5: Verify the last-vector rejection did not mutate the schema. desc = client.describe_collection(last_vector_collection_name) field_names = [field["name"] for field in desc["fields"]] assert "id" in field_names assert "vec" in field_names assert "age" in field_names for name in [collection_name, last_vector_collection_name]: self.drop_collection(client, name) @pytest.mark.tags(CaseLabel.L1) def test_drop_function_negative_constraints(self): """ TC-L1-07: Drop Function negative constraints. target: verify Drop Function rejects invalid requests and preserves schema on failure method: test empty name, missing function, detach-only BM25 rejection, repeated drop, and last-vector cascade rejection expected: errors are clear; failed drops do not partially mutate schema """ client = self._client() collection_name = f"{cf.gen_collection_name_by_testcase_name()}_function_negative" last_vector_collection_name = f"{cf.gen_collection_name_by_testcase_name()}_function_last_vector" # Step 1: Create a collection where cascade-dropping BM25 is allowed because vec remains. schema = client.create_schema(enable_dynamic_field=False, auto_id=False) schema.add_field("id", DataType.INT64, is_primary=True) schema.add_field("text", DataType.VARCHAR, max_length=1024, enable_analyzer=True) schema.add_field("sparse", DataType.SPARSE_FLOAT_VECTOR) schema.add_field("vec", DataType.FLOAT_VECTOR, dim=4) schema.add_function( Function( name="bm25", function_type=FunctionType.BM25, input_field_names=["text"], output_field_names="sparse", ) ) client.create_collection( collection_name=collection_name, schema=schema, consistency_level="Strong", ) # Step 2: Reject empty and missing function names before any successful drop. self.drop_collection_function( client, collection_name, "", check_task=ct.CheckTasks.err_res, check_items={ ct.err_code: 1100, ct.err_msg: "Must specify exactly one valid Drop identifier (drop_field_name/drop_field_id/drop_function_name)", }, ) self.drop_collection_function( client, collection_name, "missing_function", check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "function not found: missing_function: invalid parameter"}, ) desc = client.describe_collection(collection_name) assert "bm25" in [func["name"] for func in desc.get("functions", [])] assert "sparse" in [field["name"] for field in desc["fields"]] # Step 3: Reject detach-only BM25 drop and verify schema stays unchanged. self.drop_collection_function( client, collection_name, "bm25", check_task=ct.CheckTasks.err_res, check_items={ ct.err_code: 1100, ct.err_msg: ( "BM25 function must be dropped with its output field in drop_function_field interface: " "bm25: invalid parameter" ), }, ) desc = client.describe_collection(collection_name) assert "bm25" in [func["name"] for func in desc.get("functions", [])] assert "sparse" in [field["name"] for field in desc["fields"]] # Step 4: Cascade-drop BM25 once successfully, then reject repeated cascade drop. self.drop_function_field(client, collection_name, "bm25") desc = client.describe_collection(collection_name) field_names = [field["name"] for field in desc["fields"]] function_names = [func["name"] for func in desc.get("functions", [])] assert "bm25" not in function_names assert "sparse" not in field_names assert "text" in field_names assert "vec" in field_names self.drop_function_field( client, collection_name, "bm25", check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "function not found: bm25: invalid parameter"}, ) # Step 5: Create a collection where BM25 output sparse is the only vector field. last_vector_schema = client.create_schema(enable_dynamic_field=False, auto_id=False) last_vector_schema.add_field("id", DataType.INT64, is_primary=True) last_vector_schema.add_field("text", DataType.VARCHAR, max_length=1024, enable_analyzer=True) last_vector_schema.add_field("sparse", DataType.SPARSE_FLOAT_VECTOR) last_vector_schema.add_function( Function( name="bm25", function_type=FunctionType.BM25, input_field_names=["text"], output_field_names="sparse", ) ) client.create_collection( collection_name=last_vector_collection_name, schema=last_vector_schema, consistency_level="Strong", ) # Step 6: Reject cascade Drop Function because it would remove the last vector field. self.drop_function_field( client, last_vector_collection_name, "bm25", check_task=ct.CheckTasks.err_res, check_items={ ct.err_code: 1100, ct.err_msg: ( "cannot drop function bm25: it would leave no vector field in the collection: invalid parameter" ), }, ) desc = client.describe_collection(last_vector_collection_name) field_names = [field["name"] for field in desc["fields"]] function_names = [func["name"] for func in desc.get("functions", [])] assert "bm25" in function_names assert "sparse" in field_names assert "text" in field_names for name in [collection_name, last_vector_collection_name]: self.drop_collection(client, name) @pytest.mark.tags(CaseLabel.L1) def test_drop_minhash_function_detach_vs_cascade(self): """ TC-L1-07a: Drop Function detach vs cascade semantics for MinHash. target: verify MinHash supports both detach-only and cascade-output-field drop paths method: drop_collection_function keeps output field/index; drop_function_field removes output field/index expected: detach removes only the function; cascade removes function, output field, and output index """ client = self._client() detach_collection_name = f"{cf.gen_collection_name_by_testcase_name()}_detach" cascade_collection_name = f"{cf.gen_collection_name_by_testcase_name()}_cascade" def create_minhash_collection(name): schema = client.create_schema(enable_dynamic_field=False, auto_id=False) schema.add_field("id", DataType.INT64, is_primary=True) schema.add_field("doc", DataType.VARCHAR, max_length=1024) schema.add_field("mh", DataType.BINARY_VECTOR, dim=512) schema.add_field("vec", DataType.FLOAT_VECTOR, dim=4) schema.add_function( Function( name="text_to_minhash", function_type=FunctionType.MINHASH, input_field_names=["doc"], output_field_names=["mh"], params={"num_hashes": 16, "shingle_size": 3}, ) ) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") index_params.add_index( field_name="mh", index_type="MINHASH_LSH", metric_type="MHJACCARD", params={"mh_lsh_band": 8}, ) client.create_collection( collection_name=name, schema=schema, index_params=index_params, consistency_level="Strong", ) rows = [ { "id": i, "doc": f"minhash semantic document group {i % 3}", "vec": [float(i), 0.0, 0.0, 0.0], } for i in range(12) ] client.insert(collection_name=name, data=rows) client.flush(name) assert self.wait_for_index_ready(client, name, index_name="vec", timeout=120) assert self.wait_for_index_ready(client, name, index_name="mh", timeout=120) client.load_collection(name) search_res = client.search( name, data=[rows[0]["doc"]], anns_field="mh", search_params={"metric_type": "MHJACCARD", "params": {}}, limit=3, output_fields=["id", "doc"], ) assert len(search_res[0]) > 0 assert "doc" in search_res[0][0]["entity"] # Step 1: Detach-only MinHash drop removes the function but keeps output field/index. create_minhash_collection(detach_collection_name) client.drop_collection_function(detach_collection_name, "text_to_minhash") desc = client.describe_collection(detach_collection_name) field_names = [field["name"] for field in desc["fields"]] function_names = [func["name"] for func in desc.get("functions", [])] assert "text_to_minhash" not in function_names assert "mh" in field_names assert "vec" in field_names assert "mh" in client.list_indexes(detach_collection_name) # Once detached, mh is a normal field and new writes must provide it or fail clearly. self.insert( client, collection_name=detach_collection_name, data=[{"id": 100, "doc": "after detach", "vec": [1.0, 0.0, 0.0, 0.0]}], check_task=ct.CheckTasks.err_res, check_items={ ct.err_code: 1, ct.err_msg: "Insert missed an field `mh` to collection without set nullable==true or set default_value", }, ) # Step 2: Cascade MinHash drop removes the function, output field, and output index. create_minhash_collection(cascade_collection_name) self.drop_function_field(client, cascade_collection_name, "text_to_minhash") desc = client.describe_collection(cascade_collection_name) field_names = [field["name"] for field in desc["fields"]] function_names = [func["name"] for func in desc.get("functions", [])] assert "text_to_minhash" not in function_names assert "mh" not in field_names assert "doc" in field_names assert "vec" in field_names for _ in range(30): indexes = client.list_indexes(cascade_collection_name) if "mh" not in indexes: break time.sleep(1) assert "mh" not in indexes assert "vec" in indexes assert client.describe_index(cascade_collection_name, index_name="mh") is None client.insert( collection_name=cascade_collection_name, data=[{"id": 100, "doc": "after cascade", "vec": [1.0, 0.0, 0.0, 0.0]}], ) new_row = client.query(cascade_collection_name, filter="id == 100", output_fields=["*"]) assert new_row[0]["doc"] == "after cascade" assert "mh" not in new_row[0] for name in [detach_collection_name, cascade_collection_name]: self.drop_collection(client, name) @pytest.mark.tags(CaseLabel.L1) @pytest.mark.skip(reason="TC-L3-01 is stability-only; normal CI/CD cannot reliably expose in-flight DDL/DML races.") def test_drop_field_in_flight_request_semantics(self): """ TC-L3-01: In-flight request semantics around Drop Field. This placeholder keeps the original automation slot visible. The case was moved out of L1 because short CI/CD runs cannot reliably control or observe the in-flight DDL/DML timing window. """ # Intentionally empty. Run this scenario only in long-running stability jobs. pass @pytest.mark.tags(CaseLabel.L1) @pytest.mark.skip(reason="TC-L1-09 is already covered by TC-L0-01 and TC-L1-02; no duplicate automation.") def test_drop_field_sdk_schema_cache_invalidation(self): """ TC-L1-09: SDK schema cache invalidation. This placeholder keeps the checklist slot visible. The same-client post-drop describe/query/search path is covered by TC-L0-01, and same-name different-type cache correctness is covered by TC-L1-02. """ # Intentionally empty. Add a dedicated case only if a separate PyMilvus cache bug appears. pass @pytest.mark.tags(CaseLabel.L1) def test_drop_field_alias_path_schema_cache(self): """ TC-L1-10: Alias path schema cache invalidation. target: verify Drop Field cache invalidation covers both collection name and alias paths method: warm up alias describe/query/search, drop tag through alias, then verify both paths use the new schema expected: alias and collection name both hide tag; kept-field read/write paths still work """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() alias_name = f"{collection_name}_alias" # Step 1: Create a collection with a dropped scalar field and bind an alias to it. 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) schema.add_field("age", DataType.INT64) schema.add_field("tag", DataType.VARCHAR, max_length=64) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) client.create_alias(collection_name, alias_name) # Step 2: Insert data and warm up alias-side schema/read cache before drop. rows = [{"id": i, "vec": [float(i), 0.0, 0.0, 0.0], "age": 20 + i, "tag": f"tag_{i}"} for i in range(10)] client.insert(collection_name=collection_name, data=rows) client.flush(collection_name) client.load_collection(collection_name) alias_desc = client.describe_collection(alias_name) assert "tag" in [field["name"] for field in alias_desc["fields"]] alias_query_before_drop = client.query( alias_name, filter='tag == "tag_1"', output_fields=["id", "tag"], ) assert len(alias_query_before_drop) == 1 assert alias_query_before_drop[0]["tag"] == "tag_1" alias_search_before_drop = client.search( alias_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=3, output_fields=["id", "tag"], ) assert len(alias_search_before_drop[0]) == 3 assert "tag" in alias_search_before_drop[0][0]["entity"] # Step 3: Drop the field through the alias path. client.drop_collection_field(alias_name, "tag") # Step 4: Verify both collection name and alias describe paths observe the new schema. collection_desc = client.describe_collection(collection_name) alias_desc = client.describe_collection(alias_name) collection_fields = [field["name"] for field in collection_desc["fields"]] alias_fields = [field["name"] for field in alias_desc["fields"]] assert "tag" not in collection_fields assert "tag" not in alias_fields assert "age" in collection_fields assert "age" in alias_fields # Step 5: Verify kept-field query/search works through both paths. for name in [collection_name, alias_name]: query_res = client.query( name, filter="id in [0, 1, 2]", output_fields=["id", "age"], ) assert {row["id"] for row in query_res} == {0, 1, 2} assert all("tag" not in row for row in query_res) search_res = client.search( name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=3, output_fields=["id", "age"], ) assert len(search_res[0]) == 3 assert "age" in search_res[0][0]["entity"] assert "tag" not in search_res[0][0]["entity"] # Step 6: Verify explicit dropped-field references fail through both paths. for name in [collection_name, alias_name]: self.query( client, name, filter='tag == "tag_1"', output_fields=["id"], check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "field tag not exist"}, ) self.search( client, name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=3, output_fields=["id", "tag"], check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "field tag not exist"}, ) # Step 7: Verify post-drop inserts through both paths use the new schema. client.insert( collection_name=alias_name, data=[{"id": 100, "vec": [1.0, 0.0, 0.0, 0.0], "age": 100}], ) client.insert( collection_name=collection_name, data=[{"id": 101, "vec": [1.0, 1.0, 0.0, 0.0], "age": 101}], ) inserted = client.query( collection_name, filter="id in [100, 101]", output_fields=["id", "age"], ) assert {row["id"] for row in inserted} == {100, 101} assert all("tag" not in row for row in inserted) client.drop_alias(alias_name) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L2) def test_drop_vector_field_type_matrix(self): """ TC-L2-11: Drop vector field type matrix. target: verify dropping each supported vector field type does not affect the kept vector field method: create one collection with vec plus all drop vector types, then drop them one by one expected: each dropped vector field/index/search path disappears; kept vec search remains stable after reload """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() dim = 32 drop_vector_fields = [ ("drop_float_vec", DataType.FLOAT_VECTOR, "HNSW", "L2"), ("drop_binary_vec", DataType.BINARY_VECTOR, "BIN_FLAT", "HAMMING"), ("drop_float16_vec", DataType.FLOAT16_VECTOR, "HNSW", "L2"), ("drop_bfloat16_vec", DataType.BFLOAT16_VECTOR, "HNSW", "IP"), ("drop_int8_vec", DataType.INT8_VECTOR, "HNSW", "COSINE"), ("drop_sparse_vec", DataType.SPARSE_FLOAT_VECTOR, "SPARSE_INVERTED_INDEX", "IP"), ] # Step 1: Create one collection with a kept FloatVector field and all vector types to drop. 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=dim) schema.add_field("age", DataType.INT64) for field_name, vector_type, _, _ in drop_vector_fields: if vector_type == DataType.SPARSE_FLOAT_VECTOR: schema.add_field(field_name, DataType.SPARSE_FLOAT_VECTOR) else: schema.add_field(field_name, vector_type, dim=dim) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") for field_name, _, index_type, metric_type in drop_vector_fields: index_params.add_index(field_name=field_name, index_type=index_type, metric_type=metric_type) client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) # Step 2: Insert old rows that contain values for every vector field. vec_values = cf.gen_vectors(100, dim, vector_data_type=DataType.FLOAT_VECTOR) drop_values_by_field = { field_name: cf.gen_vectors(100, dim, vector_data_type=vector_type) for field_name, vector_type, _, _ in drop_vector_fields } rows = [] for i in range(100): row = { "id": i, "vec": vec_values[i], "age": 20 + i, } for field_name, _, _, _ in drop_vector_fields: row[field_name] = drop_values_by_field[field_name][i] rows.append(row) client.insert(collection_name=collection_name, data=rows) client.flush(collection_name) client.load_collection(collection_name) indexes = client.list_indexes(collection_name) assert "vec" in indexes for field_name, _, _, _ in drop_vector_fields: assert field_name in indexes assert client.describe_index(collection_name, index_name=field_name)["field_name"] == field_name # Step 3: Verify the kept vector field is searchable before dropping typed vector fields. search_before_drop = client.search( collection_name, data=[vec_values[0]], anns_field="vec", limit=5, output_fields=["id", "age"], ) assert len(search_before_drop[0]) == 5 assert "age" in search_before_drop[0][0]["entity"] # Step 4: Drop each vector type and verify the kept vector field remains searchable. for field_name, _, index_type, metric_type in drop_vector_fields: client.drop_collection_field(collection_name, field_name) desc = client.describe_collection(collection_name) field_names = [field["name"] for field in desc["fields"]] assert field_name not in field_names assert "vec" in field_names assert "age" in field_names for _ in range(30): indexes = client.list_indexes(collection_name) if field_name not in indexes: break time.sleep(1) assert field_name not in indexes assert "vec" in indexes assert client.describe_index(collection_name, index_name=field_name) is None self.search( client, collection_name, data=[drop_values_by_field[field_name][0]], anns_field=field_name, limit=5, output_fields=["id"], check_task=ct.CheckTasks.err_res, check_items={ ct.err_code: 1100, ct.err_msg: f"failed to get field schema by name: fieldName({field_name}) not found", }, ) dropped_index_params = client.prepare_index_params() dropped_index_params.add_index(field_name=field_name, index_type=index_type, metric_type=metric_type) self.create_index( client, collection_name, dropped_index_params, check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: f"cannot create index on non-exist field: {field_name}"}, ) kept_search = client.search( collection_name, data=[vec_values[0]], anns_field="vec", limit=5, output_fields=["id", "age"], ) assert len(kept_search[0]) == 5 assert "age" in kept_search[0][0]["entity"] assert field_name not in kept_search[0][0]["entity"] # Step 5: Release/load and verify old typed-vector data is still skipped. client.release_collection(collection_name) client.load_collection(collection_name) desc = client.describe_collection(collection_name) field_names = [field["name"] for field in desc["fields"]] for field_name, _, _, _ in drop_vector_fields: assert field_name not in field_names search_after_reload = client.search( collection_name, data=[vec_values[0]], anns_field="vec", limit=5, output_fields=["id", "age"], ) assert len(search_after_reload[0]) == 5 assert "age" in search_after_reload[0][0]["entity"] for field_name, _, _, _ in drop_vector_fields: assert field_name not in search_after_reload[0][0]["entity"] self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_drop_field_output_fields_wildcard_and_explicit_reference(self): """ TC-L1-12: Output fields wildcard and explicit reference after Drop Field. target: verify dropped field is invisible through wildcard output and explicit references fail method: drop tag, then query/search with ["*"], ["tag"], and ["*", "tag"] expected: wildcard output omits tag; explicit tag output fails instead of returning partial results """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Step 1: Create a collection with scalar, vector, and JSON fields. 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) schema.add_field("age", DataType.INT64) schema.add_field("tag", DataType.VARCHAR, max_length=64) schema.add_field("json_field", DataType.JSON) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) # Step 2: Insert old rows with tag and JSON data, then load. rows = [ { "id": i, "vec": [float(i), 0.0, 0.0, 0.0], "age": 20 + i, "tag": f"tag_{i}", "json_field": {"bucket": i % 2, "label": f"json_{i}"}, } for i in range(10) ] client.insert(collection_name=collection_name, data=rows) client.flush(collection_name) client.load_collection(collection_name) # Step 3: Verify wildcard output exposes tag before drop, then drop tag. before_drop = client.query(collection_name, filter="id == 1", output_fields=["*"]) assert before_drop[0]["tag"] == "tag_1" assert before_drop[0]["json_field"]["label"] == "json_1" client.drop_collection_field(collection_name, "tag") desc = client.describe_collection(collection_name) field_names = [field["name"] for field in desc["fields"]] assert "tag" not in field_names assert "json_field" in field_names assert "age" in field_names # Step 4: query(output_fields=["*"]) must not return the dropped field. query_star = client.query( collection_name, filter="id in [0, 1, 2]", output_fields=["*"], ) assert {row["id"] for row in query_star} == {0, 1, 2} for row in query_star: assert "age" in row assert "json_field" in row assert "tag" not in row # Step 5: search(output_fields=["*"]) must not return the dropped field. search_star = client.search( collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=3, output_fields=["*"], ) assert len(search_star[0]) == 3 for hit in search_star[0]: assert "age" in hit["entity"] assert "json_field" in hit["entity"] assert "tag" not in hit["entity"] # Step 6: Explicit tag output in query must fail clearly, including ["*", "tag"]. for output_fields in [["tag"], ["*", "tag"]]: self.query( client, collection_name, filter="id >= 0", output_fields=output_fields, limit=3, check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "field tag not exist"}, ) # Step 7: Explicit tag output in search must fail clearly, including ["*", "tag"]. for output_fields in [["tag"], ["*", "tag"]]: self.search( client, collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=3, output_fields=output_fields, check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "field tag not exist"}, ) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_drop_field_with_default_value(self): """ TC-L1-13: Drop field with default value. target: verify default-value fields disappear completely after Drop Field method: create default scalar fields, verify default backfill, drop them, then verify references fail expected: dropped default fields are not returned by wildcard output and new writes cannot carry them """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() default_fields = [ ("default_int", DataType.INT64, {"default_value": 42}, 42, "default_int == 42"), ("default_float", DataType.FLOAT, {"default_value": 1.5}, 1.5, "default_float == 1.5"), ("default_bool", DataType.BOOL, {"default_value": True}, True, "default_bool == true"), ( "default_varchar", DataType.VARCHAR, {"max_length": 64, "default_value": "default"}, "default", 'default_varchar == "default"', ), ] # Step 1: Create a collection with supported scalar default-value fields. 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) schema.add_field("age", DataType.INT64) for field_name, field_type, field_kwargs, _, _ in default_fields: schema.add_field(field_name, field_type, **field_kwargs) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) # Step 2: Insert rows that omit default fields and verify default backfill before drop. rows = [{"id": i, "vec": [float(i), 0.0, 0.0, 0.0], "age": 20 + i} for i in range(10)] client.insert(collection_name=collection_name, data=rows) client.flush(collection_name) client.load_collection(collection_name) before_drop = client.query(collection_name, filter="id == 1", output_fields=["*"])[0] for field_name, _, _, expected_value, _ in default_fields: assert before_drop[field_name] == expected_value # Step 3: Drop every default-value field. for field_name, _, _, _, _ in default_fields: client.drop_collection_field(collection_name, field_name) # Step 4: Verify dropped fields are no longer visible in schema. field_names = [field["name"] for field in client.describe_collection(collection_name)["fields"]] for field_name, _, _, _, _ in default_fields: assert field_name not in field_names assert "id" in field_names assert "vec" in field_names assert "age" in field_names # Step 5: Wildcard query/search output must not return dropped default fields. query_star = client.query(collection_name, filter="id in [0, 1, 2]", output_fields=["*"]) for row in query_star: assert "age" in row for field_name, _, _, _, _ in default_fields: assert field_name not in row search_star = client.search( collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=3, output_fields=["*"], ) assert len(search_star[0]) == 3 for hit in search_star[0]: assert "age" in hit["entity"] for field_name, _, _, _, _ in default_fields: assert field_name not in hit["entity"] # Step 6: Explicit output/filter references to dropped default fields must fail. for field_name, _, _, _, filter_expr in default_fields: self.query( client, collection_name, filter="id >= 0", output_fields=[field_name], limit=1, check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: f"field {field_name} not exist"}, ) self.query( client, collection_name, filter=filter_expr, output_fields=["id"], limit=1, check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: f"field {field_name} not exist"}, ) # Step 7: New insert without dropped fields succeeds; insert carrying a dropped field fails. client.insert( collection_name=collection_name, data=[{"id": 100, "vec": [1.0, 0.0, 0.0, 0.0], "age": 100}], ) client.flush(collection_name) new_row = client.query(collection_name, filter="id == 100", output_fields=["*"])[0] assert new_row["age"] == 100 for field_name, _, _, _, _ in default_fields: assert field_name not in new_row self.insert( client, collection_name=collection_name, data=[{"id": 101, "vec": [1.0, 0.0, 0.0, 0.0], "age": 101, "default_int": 42}], check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1, ct.err_msg: "Attempt to insert an unexpected field `default_int`"}, ) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) @pytest.mark.xfail( reason="Known issue: https://github.com/milvus-io/milvus/issues/50484", strict=False, ) def test_drop_then_add_same_name_analyzer_field(self): """ TC-L1-14: Drop analyzer field and re-add same-name field with different analyzer params. target: verify old analyzer tokens/index data do not pollute a re-added same-name analyzer field method: drop text_content with standard analyzer, re-add text_content with length filter, then query text_match expected: old long-token rows are not matched; new analyzer semantics only apply to post-add rows """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() old_token = "legacytoken" new_token = "new" old_analyzer_params = {"tokenizer": "standard"} # The new analyzer drops tokens longer than 4 chars. This makes legacytoken searchable before drop # but unsearchable after re-add, so any post-readd match would indicate old analyzer/index leakage. new_analyzer_params = { "tokenizer": "standard", "filter": [ { "type": "length", "max": 4, } ], } # Step 1: Create text_content with the old analyzer and an inverted index. 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) schema.add_field("age", DataType.INT64) schema.add_field( "text_content", DataType.VARCHAR, max_length=1024, enable_analyzer=True, enable_match=True, analyzer_params=old_analyzer_params, ) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") index_params.add_index(field_name="text_content", index_type="INVERTED") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) # Step 2: Insert old rows and verify the old analyzer can match the long token. old_rows = [ { "id": i, "vec": [float(i), 0.0, 0.0, 0.0], "age": 20 + i, "text_content": f"{old_token} old document {i}", } for i in range(5) ] client.insert(collection_name=collection_name, data=old_rows) client.flush(collection_name) client.load_collection(collection_name) assert self.wait_for_index_ready(client, collection_name, index_name="vec", timeout=60) assert self.wait_for_index_ready(client, collection_name, index_name="text_content", timeout=60) before_drop_desc = client.describe_collection(collection_name) old_text_field = next(field for field in before_drop_desc["fields"] if field["name"] == "text_content") old_text_field_id = old_text_field.get("field_id") old_match = client.query( collection_name, filter=f"text_match(text_content, '{old_token}')", output_fields=["id", "text_content"], ) assert {row["id"] for row in old_match} == {0, 1, 2, 3, 4} assert all(old_token in row["text_content"] for row in old_match) # Step 3: Drop text_content and wait until its index metadata is removed. client.drop_collection_field(collection_name, "text_content") after_drop_desc = client.describe_collection(collection_name) field_names = [field["name"] for field in after_drop_desc["fields"]] assert "text_content" not in field_names # Record the field-id high watermark when the SDK exposes it. This is a semi-whitebox guard # against reusing the old dropped field id; text_match checks below are the primary validation. max_field_id_after_drop = None properties = after_drop_desc.get("properties", {}) if isinstance(properties, dict) and properties.get("max_field_id") is not None: max_field_id_after_drop = int(properties["max_field_id"]) elif isinstance(properties, list): for prop in properties: if prop.get("key") == "max_field_id": max_field_id_after_drop = int(prop["value"]) break for _ in range(30): indexes = client.list_indexes(collection_name) if "text_content" not in indexes: break time.sleep(1) assert "text_content" not in indexes assert client.describe_index(collection_name, index_name="text_content") is None self.query( client, collection_name, filter=f"text_match(text_content, '{old_token}')", output_fields=["id"], check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "field text_content not exist"}, ) # Step 4: Re-add same-name text_content with a new analyzer that drops tokens longer than 4 chars. client.add_collection_field( collection_name, field_name="text_content", data_type=DataType.VARCHAR, nullable=True, max_length=1024, enable_analyzer=True, enable_match=True, analyzer_params=new_analyzer_params, ) after_add_desc = client.describe_collection(collection_name) new_text_field = next(field for field in after_add_desc["fields"] if field["name"] == "text_content") new_text_field_id = new_text_field.get("field_id") if old_text_field_id is not None and new_text_field_id is not None: assert new_text_field_id > old_text_field_id if max_field_id_after_drop is not None and new_text_field_id is not None: assert new_text_field_id > max_field_id_after_drop new_index_params = client.prepare_index_params() new_index_params.add_index(field_name="text_content", index_type="INVERTED") client.create_index(collection_name, new_index_params) assert self.wait_for_index_ready(client, collection_name, index_name="text_content", timeout=60) # Step 5: Insert post-add rows and verify only new analyzer semantics are visible. new_rows = [ { "id": 100, "vec": [1.0, 0.0, 0.0, 0.0], "age": 100, "text_content": f"{new_token} {old_token}", }, { "id": 101, "vec": [2.0, 0.0, 0.0, 0.0], "age": 101, "text_content": "tiny new word", }, ] client.insert(collection_name=collection_name, data=new_rows) client.flush(collection_name) new_token_match = client.query( collection_name, filter=f"text_match(text_content, '{new_token}')", output_fields=["id", "text_content"], ) assert {row["id"] for row in new_token_match} == {100, 101} old_token_after_readd = client.query( collection_name, filter=f"text_match(text_content, '{old_token}')", output_fields=["id", "text_content"], ) assert len(old_token_after_readd) == 0 old_rows_after_readd = client.query( collection_name, filter="id in [0, 1, 2, 3, 4]", output_fields=["id", "text_content"], ) assert len(old_rows_after_readd) == 5 for row in old_rows_after_readd: assert row.get("text_content") is None or old_token not in row.get("text_content", "") # Step 6: Reload and repeat the key old/new analyzer checks. client.release_collection(collection_name) client.load_collection(collection_name) new_token_after_reload = client.query( collection_name, filter=f"text_match(text_content, '{new_token}')", output_fields=["id", "text_content"], ) assert {row["id"] for row in new_token_after_reload} == {100, 101} old_token_after_reload = client.query( collection_name, filter=f"text_match(text_content, '{old_token}')", output_fields=["id", "text_content"], ) assert len(old_token_after_reload) == 0 search_res = client.search( collection_name, data=[[1.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=3, output_fields=["id", "age", "text_content"], ) assert len(search_res[0]) == 3 for hit in search_res[0]: if hit["entity"]["id"] < 100: assert hit["entity"].get("text_content") is None self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_drop_field_upsert_and_partial_update_paths(self): """ TC-L1-15: Upsert and partial update after Drop Field. target: verify update paths do not revive dropped field data after schema evolution method: drop score, then run upsert and upsert(partial_update=True) with and without score expected: kept-field updates work; dropped score is rejected without dynamic schema and becomes dynamic data with it """ client = self._client() static_collection_name = f"{cf.gen_collection_name_by_testcase_name()}_static" dynamic_collection_name = f"{cf.gen_collection_name_by_testcase_name()}_dynamic" # Step 1: Create a non-dynamic collection with score as a normal scalar field. 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) schema.add_field("age", DataType.INT64) schema.add_field("score", DataType.INT64) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=static_collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) rows = [ {"id": 0, "vec": [0.0, 0.0, 0.0, 0.0], "age": 20, "score": 100}, {"id": 1, "vec": [1.0, 0.0, 0.0, 0.0], "age": 21, "score": 101}, {"id": 2, "vec": [2.0, 0.0, 0.0, 0.0], "age": 22, "score": 102}, ] client.insert(collection_name=static_collection_name, data=rows) client.flush(static_collection_name) client.load_collection(static_collection_name) before_drop = client.query(static_collection_name, filter="id in [0, 1, 2]", output_fields=["*"]) assert {row["score"] for row in before_drop} == {100, 101, 102} # Step 2: Drop score. From this point on, score must no longer be a schema field. client.drop_collection_field(static_collection_name, "score") field_names = [field["name"] for field in client.describe_collection(static_collection_name)["fields"]] assert "score" not in field_names # Verify full upsert can replace an old row using only kept fields. # This must not bring the old score value back through wildcard output. client.upsert( collection_name=static_collection_name, data=[{"id": 0, "vec": [0.5, 0.0, 0.0, 0.0], "age": 200}], ) row0 = client.query(static_collection_name, filter="id == 0", output_fields=["*"])[0] assert row0["age"] == 200 assert "score" not in row0 # Verify partial_update updates only the provided kept field and preserves other kept fields. # The dropped score field must still stay invisible. client.upsert( collection_name=static_collection_name, data=[{"id": 1, "age": 201}], partial_update=True, ) row1 = client.query(static_collection_name, filter="id == 1", output_fields=["*"])[0] assert row1["age"] == 201 assert row1["vec"] == [1.0, 0.0, 0.0, 0.0] assert "score" not in row1 # Step 3: Non-dynamic collection rejects dropped score on both update paths. for kwargs in [ { "data": [{"id": 0, "vec": [0.6, 0.0, 0.0, 0.0], "age": 300, "score": 999}], }, { "data": [{"id": 1, "score": 998}], "partial_update": True, }, ]: self.upsert( client, collection_name=static_collection_name, check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1, ct.err_msg: "Attempt to insert an unexpected field `score`"}, **kwargs, ) search_res = client.search( static_collection_name, data=[[0.5, 0.0, 0.0, 0.0]], anns_field="vec", limit=3, output_fields=["id", "age"], ) assert len(search_res[0]) == 3 assert "score" not in search_res[0][0]["entity"] # Step 4: Repeat with dynamic schema enabled to distinguish dynamic keys from revived schema fields. dynamic_schema = client.create_schema(enable_dynamic_field=True, auto_id=False) dynamic_schema.add_field("id", DataType.INT64, is_primary=True) dynamic_schema.add_field("vec", DataType.FLOAT_VECTOR, dim=4) dynamic_schema.add_field("age", DataType.INT64) dynamic_schema.add_field("score", DataType.INT64) dynamic_index_params = client.prepare_index_params() dynamic_index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=dynamic_collection_name, schema=dynamic_schema, index_params=dynamic_index_params, consistency_level="Strong", ) client.insert(collection_name=dynamic_collection_name, data=rows) client.flush(dynamic_collection_name) client.load_collection(dynamic_collection_name) dynamic_before_drop = client.query(dynamic_collection_name, filter="id in [0, 1, 2]", output_fields=["*"]) assert {row["score"] for row in dynamic_before_drop} == {100, 101, 102} client.drop_collection_field(dynamic_collection_name, "score") dynamic_field_names = [field["name"] for field in client.describe_collection(dynamic_collection_name)["fields"]] assert "score" not in dynamic_field_names # Verify kept-field full upsert works and does not expose the old schema score value. client.upsert( collection_name=dynamic_collection_name, data=[{"id": 0, "vec": [0.5, 0.0, 0.0, 0.0], "age": 200}], ) dynamic_row0 = client.query(dynamic_collection_name, filter="id == 0", output_fields=["*"])[0] assert dynamic_row0["age"] == 200 assert "score" not in dynamic_row0 # Verify kept-field partial_update preserves the vector and still does not expose old score. client.upsert( collection_name=dynamic_collection_name, data=[{"id": 1, "age": 201}], partial_update=True, ) dynamic_row1 = client.query(dynamic_collection_name, filter="id == 1", output_fields=["*"])[0] assert dynamic_row1["age"] == 201 assert dynamic_row1["vec"] == [1.0, 0.0, 0.0, 0.0] assert "score" not in dynamic_row1 # id=2 is never rewritten with score after drop; this sentinel proves old score=102 does not revive. dynamic_row2 = client.query(dynamic_collection_name, filter="id == 2", output_fields=["*"])[0] assert dynamic_row2["age"] == 22 assert "score" not in dynamic_row2 # Step 5: With dynamic schema enabled, post-drop score is accepted only as newly written dynamic data. client.upsert( collection_name=dynamic_collection_name, data=[{"id": 0, "vec": [0.6, 0.0, 0.0, 0.0], "age": 300, "score": 999}], ) dynamic_score_row0 = client.query(dynamic_collection_name, filter="id == 0", output_fields=["*"])[0] assert dynamic_score_row0["age"] == 300 assert dynamic_score_row0["score"] == 999 assert dynamic_score_row0["score"] != 100 client.upsert( collection_name=dynamic_collection_name, data=[{"id": 1, "score": 998}], partial_update=True, ) dynamic_score_row1 = client.query(dynamic_collection_name, filter="id == 1", output_fields=["*"])[0] assert dynamic_score_row1["age"] == 201 assert dynamic_score_row1["score"] == 998 assert dynamic_score_row1["score"] != 101 dynamic_row2_after_dynamic_writes = client.query( dynamic_collection_name, filter="id == 2", output_fields=["*"], )[0] assert "score" not in dynamic_row2_after_dynamic_writes # Step 6: The schema still has no score field; returned score values above are dynamic keys after drop. dynamic_field_names = [field["name"] for field in client.describe_collection(dynamic_collection_name)["fields"]] assert "score" not in dynamic_field_names self.drop_collection(client, static_collection_name) self.drop_collection(client, dynamic_collection_name) @pytest.mark.tags(CaseLabel.L2) def test_drop_field_query_and_search_iterator_paths(self): """ TC-L2-16: Query/search iterator after Drop Field. target: verify iterator paths use the post-drop schema without losing iterator snapshot isolation method: drop tag while query_iterator is open, insert a new row, then continue iterator and run post-drop iterators expected: existing iterator omits tag after drop and does not see the new row; tag references fail clearly """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Step 1: Create a collection and insert multiple batches so iterator pagination is exercised. 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) schema.add_field("age", DataType.INT64) schema.add_field("tag", DataType.VARCHAR, max_length=64) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) rows = [] for batch_id in range(3): batch_rows = [ { "id": batch_id * 10 + i, "vec": [float(batch_id * 10 + i), 0.0, 0.0, 0.0], "age": 20 + batch_id * 10 + i, "tag": f"tag_{batch_id}_{i}", } for i in range(10) ] client.insert(collection_name=collection_name, data=batch_rows) rows.extend(batch_rows) client.flush(collection_name) client.load_collection(collection_name) # Step 2: Start query_iterator before drop and consume one batch under the old schema. query_iterator = client.query_iterator( collection_name=collection_name, batch_size=5, limit=100, filter="id >= 0", output_fields=["*"], ) first_batch = query_iterator.next() assert len(first_batch) == 5 first_batch_ids = {row["id"] for row in first_batch} assert len(first_batch_ids) == len(first_batch) assert first_batch_ids.issubset({row["id"] for row in rows}) for row in first_batch: assert "tag" in row # Step 3: Drop tag and insert a new row after the iterator snapshot was established. client.drop_collection_field(collection_name, "tag") field_names = [field["name"] for field in client.describe_collection(collection_name)["fields"]] assert "tag" not in field_names client.insert( collection_name=collection_name, data=[{"id": 999, "vec": [999.0, 0.0, 0.0, 0.0], "age": 999}], ) client.flush(collection_name) # A fresh query sees the new row, so the iterator check below proves snapshot isolation. new_row = client.query( collection_name=collection_name, filter="id == 999", output_fields=["*"], ) assert len(new_row) == 1 assert new_row[0]["id"] == 999 assert "tag" not in new_row[0] # Step 4: Continue the old query_iterator. It uses the new schema, but not the new data snapshot. query_rows_after_drop = [] while True: batch = query_iterator.next() if len(batch) == 0: query_iterator.close() break query_rows_after_drop.extend(batch) remaining_ids = {row["id"] for row in query_rows_after_drop} expected_remaining_ids = {row["id"] for row in rows} - first_batch_ids assert len(query_rows_after_drop) == len(expected_remaining_ids) assert len(remaining_ids) == len(query_rows_after_drop) assert remaining_ids == expected_remaining_ids assert 999 not in remaining_ids for row in query_rows_after_drop: assert "age" in row assert "tag" not in row # Step 5: search_iterator created after drop must use the new schema for every hit. search_hits = [] search_iterator = client.search_iterator( collection_name=collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", batch_size=5, limit=12, output_fields=["*"], ) while True: batch = search_iterator.next() if not batch: search_iterator.close() break search_hits.extend(batch) assert len(search_hits) == 12 for hit in search_hits: assert "age" in hit["entity"] assert "tag" not in hit["entity"] # Step 6: iterator requests that explicitly output the dropped field must fail. query_output_iterator = client.query_iterator( collection_name=collection_name, batch_size=3, limit=3, filter="id >= 0", output_fields=["tag"], ) with pytest.raises(Exception) as query_output_exc: query_output_iterator.next() query_output_iterator.close() assert "field tag not exist" in str(query_output_exc.value).lower(), str(query_output_exc.value) with pytest.raises(Exception) as search_output_exc: client.search_iterator( collection_name=collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", batch_size=3, limit=3, output_fields=["tag"], ) assert "field tag not exist" in str(search_output_exc.value).lower(), str(search_output_exc.value) # Step 7: iterator filters that explicitly reference the dropped field must also fail. with pytest.raises(Exception) as query_filter_exc: client.query_iterator( collection_name=collection_name, batch_size=3, limit=3, filter='tag == "tag_0_0"', output_fields=["id"], ) assert "field tag not exist" in str(query_filter_exc.value).lower(), str(query_filter_exc.value) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_drop_field_runtime_query_reference_paths(self): """ TC-L1-17: Drop field referenced by runtime query options. target: verify runtime query options fail clearly after their referenced fields are dropped method: drop fields referenced by decay reranker, group_by_field, and order_by_fields expected: Drop Field succeeds; post-drop runtime references return field-not-found errors """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Step 1: Create one collection with independent fields for each runtime reference path. 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) schema.add_field("rank_score", DataType.INT64) schema.add_field("group_field", DataType.VARCHAR, max_length=64) schema.add_field("order_field", DataType.FLOAT) schema.add_field("age", DataType.INT64) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) rows = [ { "id": i, "vec": [float(i), 0.0, 0.0, 0.0], "rank_score": i * 10, "group_field": f"group_{i % 3}", "order_field": float(30 - i), "age": 20 + i, } for i in range(12) ] client.insert(collection_name=collection_name, data=rows) client.flush(collection_name) client.load_collection(collection_name) decay_ranker = Function( name="rank_score_decay", input_field_names=["rank_score"], function_type=FunctionType.RERANK, params={"reranker": "decay", "function": "gauss", "origin": 0, "offset": 0, "decay": 0.5, "scale": 100}, ) # Step 2: Verify the three runtime reference paths work before Drop Field. rerank_res = client.search( collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", ranker=decay_ranker, limit=5, output_fields=["id", "rank_score"], ) assert len(rerank_res[0]) == 5 assert "rank_score" in rerank_res[0][0]["entity"] group_res = client.search( collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", group_by_field="group_field", limit=3, output_fields=["id", "group_field"], ) assert len(group_res[0]) == 3 group_values = [hit["entity"]["group_field"] for hit in group_res[0]] assert len(group_values) == len(set(group_values)) order_res = client.search( collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=5, output_fields=["id", "order_field"], order_by_fields=[{"field": "order_field", "order": "asc"}], ) assert len(order_res[0]) == 5 order_values = [hit["entity"]["order_field"] for hit in order_res[0]] assert order_values == sorted(order_values) # Step 3: Drop the reranker input. Drop Field is allowed because this is a request-time dependency. client.drop_collection_field(collection_name, "rank_score") field_names = [field["name"] for field in client.describe_collection(collection_name)["fields"]] assert "rank_score" not in field_names self.search( client, collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", ranker=decay_ranker, limit=5, output_fields=["id"], check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "input field rank_score not found in collection schema"}, ) # Step 4: Drop the group-by field. Post-drop group_by_field must fail clearly. client.drop_collection_field(collection_name, "group_field") field_names = [field["name"] for field in client.describe_collection(collection_name)["fields"]] assert "group_field" not in field_names self.search( client, collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", group_by_field="group_field", limit=3, output_fields=["id"], check_task=ct.CheckTasks.err_res, check_items={ ct.err_code: 1100, ct.err_msg: "groupBy field not found in schema: field not found[field=group_field]", }, ) # Step 5: Drop the order-by field. Post-drop order_by_fields must fail clearly. client.drop_collection_field(collection_name, "order_field") field_names = [field["name"] for field in client.describe_collection(collection_name)["fields"]] assert "order_field" not in field_names self.search( client, collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=5, output_fields=["id"], order_by_fields=[{"field": "order_field", "order": "asc"}], check_task=ct.CheckTasks.err_res, check_items={ ct.err_code: 1100, ct.err_msg: "order_by field 'order_field' does not exist in collection schema: invalid parameter", }, ) # Step 6: Kept fields remain queryable after all runtime dependency fields are dropped. kept_res = client.search( collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=3, output_fields=["id", "age"], ) assert len(kept_res[0]) == 3 assert "age" in kept_res[0][0]["entity"] assert "rank_score" not in kept_res[0][0]["entity"] assert "group_field" not in kept_res[0][0]["entity"] assert "order_field" not in kept_res[0][0]["entity"] self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_drop_timestamptz_field(self): """ TC-L1-18: Drop TIMESTAMPTZ field. target: verify Drop Field works for a normal TIMESTAMPTZ field method: query/search by TIMESTAMPTZ before drop, then drop it and retry references expected: TIMESTAMPTZ field is removed; post-drop references fail; other fields still work """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Step 1: Create a collection with a normal TIMESTAMPTZ field. 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) schema.add_field("event_time", DataType.TIMESTAMPTZ, nullable=True) schema.add_field("age", DataType.INT64) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) rows = [ {"id": 0, "vec": [0.0, 0.0, 0.0, 0.0], "event_time": "2025-01-01T00:00:00Z", "age": 20}, {"id": 1, "vec": [1.0, 0.0, 0.0, 0.0], "event_time": "2025-01-02T00:00:00Z", "age": 21}, {"id": 2, "vec": [2.0, 0.0, 0.0, 0.0], "event_time": "2025-01-03T00:00:00Z", "age": 22}, ] client.insert(collection_name=collection_name, data=rows) client.flush(collection_name) client.load_collection(collection_name) # Step 2: Verify TIMESTAMPTZ filter and output work before Drop Field. time_filter = "event_time >= ISO '2025-01-02T00:00:00Z'" query_res = client.query( collection_name=collection_name, filter=time_filter, output_fields=["id", "event_time"], ) assert {row["id"] for row in query_res} == {1, 2} for row in query_res: assert "event_time" in row search_res = client.search( collection_name=collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", filter=time_filter, limit=2, output_fields=["id", "event_time"], ) assert len(search_res[0]) == 2 assert {hit["entity"]["id"] for hit in search_res[0]} == {1, 2} for hit in search_res[0]: assert "event_time" in hit["entity"] # Step 3: Drop the TIMESTAMPTZ field and verify schema convergence. client.drop_collection_field(collection_name, "event_time") field_names = [field["name"] for field in client.describe_collection(collection_name)["fields"]] assert "event_time" not in field_names assert "age" in field_names assert "vec" in field_names # Step 4: Explicit post-drop references to TIMESTAMPTZ output must fail. self.query( client, collection_name=collection_name, filter="id >= 0", output_fields=["event_time"], check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "field event_time not exist"}, ) self.search( client, collection_name=collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=2, output_fields=["event_time"], check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "field event_time not exist"}, ) # Step 5: Explicit post-drop TIMESTAMPTZ filters must fail. # Use a field-existence predicate after drop; ISO literals require the field type to exist during parsing. post_drop_time_filter = "event_time is not null" self.query( client, collection_name=collection_name, filter=post_drop_time_filter, output_fields=["id"], check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "field event_time not exist"}, ) self.search( client, collection_name=collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", filter=post_drop_time_filter, limit=2, output_fields=["id"], check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "field event_time not exist"}, ) # Step 6: Query/search paths that do not reference the dropped field still work. post_drop_query = client.query( collection_name=collection_name, filter="id in [0, 1, 2]", output_fields=["*"], ) assert {row["id"] for row in post_drop_query} == {0, 1, 2} for row in post_drop_query: assert "age" in row assert "event_time" not in row post_drop_search = client.search( collection_name=collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=3, output_fields=["*"], ) assert len(post_drop_search[0]) == 3 for hit in post_drop_search[0]: assert "age" in hit["entity"] assert "event_time" not in hit["entity"] self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_drop_active_ttl_field_rejected_until_unbound(self): """ TC-L1-18a: Reject Drop active entity TTL field. target: verify an active ttl_field cannot be dropped until the TTL binding is removed method: create ttl TIMESTAMPTZ with properties.ttl_field, reject drop, remove property, then drop expected: active drop does not mutate schema/properties; unbound TIMESTAMPTZ field can be dropped """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Step 1: Create an entity-level TTL collection whose active TTL field is `ttl`. schema = client.create_schema(enable_dynamic_field=False, auto_id=False) schema.add_field("id", DataType.INT64, is_primary=True) schema.add_field("ttl", DataType.TIMESTAMPTZ, nullable=True) schema.add_field("vec", DataType.FLOAT_VECTOR, dim=4) schema.add_field("age", DataType.INT64) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, properties={"ttl_field": "ttl", "timezone": "UTC"}, consistency_level="Strong", ) client.insert( collection_name=collection_name, data=[ { "id": 1, "ttl": "2099-01-01T00:00:00Z", "vec": [0.0, 0.0, 0.0, 0.0], "age": 10, } ], ) client.flush(collection_name) client.load_collection(collection_name) before_query = client.query(collection_name, filter="id == 1", output_fields=["id", "ttl", "age"]) assert len(before_query) == 1 assert before_query[0]["age"] == 10 before_desc = client.describe_collection(collection_name) assert before_desc.get("properties", {}).get("ttl_field") == "ttl" assert "ttl" in [field["name"] for field in before_desc["fields"]] # Step 2: Reject dropping the active TTL field and verify no metadata is partially mutated. self.drop_collection_field( client, collection_name, field_name="ttl", check_task=ct.CheckTasks.err_res, check_items={ ct.err_code: 1100, ct.err_msg: ( "cannot drop field ttl because it is referenced by collection property ttl_field, " "drop the property first" ), }, ) after_reject_desc = client.describe_collection(collection_name) assert after_reject_desc.get("properties", {}).get("ttl_field") == "ttl" assert "ttl" in [field["name"] for field in after_reject_desc["fields"]] still_visible = client.query(collection_name, filter="id == 1", output_fields=["id", "ttl", "age"]) assert len(still_visible) == 1 assert still_visible[0]["age"] == 10 self.add_collection_field( client, collection_name, field_name="ttl", data_type=DataType.INT64, nullable=True, default_value=0, check_task=ct.CheckTasks.err_res, check_items={ ct.err_code: 1100, ct.err_msg: "duplicated field name ttl", }, ) # Step 3: Explicitly remove the TTL binding; the same TIMESTAMPTZ field is now a normal droppable field. client.drop_collection_properties(collection_name, property_keys=["ttl_field"]) unbound_desc = client.describe_collection(collection_name) assert "ttl_field" not in unbound_desc.get("properties", {}) assert "ttl" in [field["name"] for field in unbound_desc["fields"]] client.drop_collection_field(collection_name, "ttl") after_drop_desc = client.describe_collection(collection_name) assert "ttl_field" not in after_drop_desc.get("properties", {}) assert "ttl" not in [field["name"] for field in after_drop_desc["fields"]] after_drop_query = client.query(collection_name, filter="id == 1", output_fields=["id", "age"]) assert len(after_drop_query) == 1 assert after_drop_query[0]["age"] == 10 assert "ttl" not in after_drop_query[0] search_res = client.search( collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=1, output_fields=["id", "age"], ) assert len(search_res[0]) == 1 assert search_res[0][0]["entity"]["id"] == 1 assert "ttl" not in search_res[0][0]["entity"] self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L2) def test_drop_loaded_field_after_partial_load(self): """ TC-L2-18b: Drop a loaded field after partial load. target: verify partial-load metadata does not keep a dropped field reference method: load selected fields including drop_me, drop it, reload, then add same-name different-type field expected: stale load_fields metadata does not break reload/search/query or bind to the new field """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Step 1: Create a collection with two scalar fields: # keep is the control field; drop_me is the field recorded in load_fields and then dropped. 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) schema.add_field("keep", DataType.VARCHAR, max_length=64) schema.add_field("drop_me", DataType.VARCHAR, max_length=64) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) # Step 2: Insert sealed rows so partial load and reload both read persisted segment data. rows = [ { "id": i, "vec": [float(i), 0.0, 0.0, 0.0], "keep": f"keep_{i}", "drop_me": f"old_drop_{i}", } for i in range(5) ] client.insert(collection_name=collection_name, data=rows) client.flush(collection_name) # Step 3: Partial-load the collection with drop_me explicitly included in the load field list. client.load_collection(collection_name, load_fields=["id", "vec", "keep", "drop_me"]) # Verify the old load_fields list is effective before Drop Field: both keep and drop_me are readable. before_drop = client.query( collection_name=collection_name, filter="id in [0, 1, 2, 3, 4]", output_fields=["id", "keep", "drop_me"], ) assert len(before_drop) == 5 for row in before_drop: assert row["keep"] == f"keep_{row['id']}" assert row["drop_me"] == f"old_drop_{row['id']}" # Step 4: Drop the loaded field and verify the schema no longer exposes it. client.drop_collection_field(collection_name, "drop_me") # Verify schema metadata has removed the dropped field while preserving the remaining fields. desc = client.describe_collection(collection_name) field_names = [field["name"] for field in desc["fields"]] assert "drop_me" not in field_names assert "keep" in field_names assert "vec" in field_names # Step 5: Verify operations that use only remaining loaded fields still work after the drop. post_drop_query = client.query( collection_name=collection_name, filter="id in [0, 1, 2, 3, 4]", output_fields=["id", "keep"], ) assert len(post_drop_query) == 5 assert {row["keep"] for row in post_drop_query} == {f"keep_{i}" for i in range(5)} post_drop_search = client.search( collection_name=collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=3, output_fields=["id", "keep"], ) assert len(post_drop_search[0]) == 3 assert "keep" in post_drop_search[0][0]["entity"] assert "drop_me" not in post_drop_search[0][0]["entity"] # Step 6: Verify explicit references to drop_me fail even though old load_fields contained it. self.query( client, collection_name=collection_name, filter="id >= 0", output_fields=["drop_me"], check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "field drop_me not exist"}, ) self.query( client, collection_name=collection_name, filter='drop_me == "old_drop_1"', output_fields=["id"], check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "field drop_me not exist"}, ) # Step 7: Release and reload without explicit load_fields. # This validates QueryCoord/QueryNode do not require the stale old field ID from the previous partial load. client.release_collection(collection_name) client.load_collection(collection_name) # Verify wildcard query/search follows the latest schema after reload and does not expose drop_me. after_reload = client.query( collection_name=collection_name, filter="id in [0, 1, 2, 3, 4]", output_fields=["*"], ) assert len(after_reload) == 5 for row in after_reload: assert "keep" in row assert "drop_me" not in row after_reload_search = client.search( collection_name=collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=3, output_fields=["*"], ) assert len(after_reload_search[0]) == 3 assert "keep" in after_reload_search[0][0]["entity"] assert "drop_me" not in after_reload_search[0][0]["entity"] # Step 8: Re-add the same field name with a different type. # This verifies stale load_fields metadata is not rebound to the new same-name field. client.add_collection_field( collection_name, field_name="drop_me", data_type=DataType.INT64, nullable=True, ) client.insert( collection_name=collection_name, data=[{"id": 100, "vec": [1.0, 0.0, 0.0, 0.0], "keep": "keep_new", "drop_me": 9000}], ) client.flush(collection_name) client.release_collection(collection_name) client.load_collection(collection_name) # Verify the new Int64 drop_me field is queryable with new data. new_field_query = client.query( collection_name=collection_name, filter="drop_me >= 9000", output_fields=["id", "drop_me"], ) assert len(new_field_query) == 1 assert new_field_query[0]["id"] == 100 assert new_field_query[0]["drop_me"] == 9000 # Verify old rows do not expose the old VarChar drop_me values through the new field. old_rows_after_readd = client.query( collection_name=collection_name, filter="id in [0, 1, 2, 3, 4]", output_fields=["id", "drop_me"], ) assert len(old_rows_after_readd) == 5 for row in old_rows_after_readd: assert row.get("drop_me") != f"old_drop_{row['id']}" self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_drop_field_then_truncate_then_write(self): """ TC-L1-19: Truncate then write after Drop Field. target: verify truncate keeps the latest schema after a field is dropped method: drop tag, truncate old data, insert new rows without tag, then validate reads and rejected tag references expected: truncate clears data only; it does not restore the dropped field or accept old-schema writes """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Step 1: Create a collection with a scalar field that will be dropped before truncate. 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) schema.add_field("age", DataType.INT64) schema.add_field("tag", DataType.VARCHAR, max_length=64) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) # Step 2: Insert old rows with tag and confirm the old schema path works before Drop Field. old_rows = [{"id": i, "vec": [float(i), 0.0, 0.0, 0.0], "age": 20 + i, "tag": f"old_tag_{i}"} for i in range(5)] client.insert(collection_name=collection_name, data=old_rows) client.flush(collection_name) client.load_collection(collection_name) before_drop = client.query( collection_name=collection_name, filter='tag == "old_tag_1"', output_fields=["id", "tag"], ) assert len(before_drop) == 1 assert before_drop[0]["tag"] == "old_tag_1" # Step 3: Drop tag and verify the schema has moved to the new shape before truncate. client.drop_collection_field(collection_name, "tag") field_names = [field["name"] for field in client.describe_collection(collection_name)["fields"]] assert "tag" not in field_names assert "age" in field_names assert "vec" in field_names # Step 4: Truncate after Drop Field; this should clear rows without restoring the dropped field. self.truncate_collection(client, collection_name) count_after_truncate = client.query( collection_name=collection_name, filter="id >= 0", output_fields=["count(*)"], ) assert count_after_truncate[0]["count(*)"] == 0 # Step 5: Insert new rows using only the latest schema. new_rows = [{"id": 100 + i, "vec": [float(i), 1.0, 0.0, 0.0], "age": 100 + i} for i in range(3)] client.insert(collection_name=collection_name, data=new_rows) client.flush(collection_name) # Step 6: Wildcard query/search should return only fields from the latest schema. query_res = client.query( collection_name=collection_name, filter="id in [100, 101, 102]", output_fields=["*"], ) assert len(query_res) == 3 for row in query_res: assert "age" in row assert "tag" not in row search_res = client.search( collection_name=collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=3, output_fields=["*"], ) assert len(search_res[0]) == 3 for hit in search_res[0]: assert "age" in hit["entity"] assert "tag" not in hit["entity"] # Step 7: Old-schema writes and explicit tag reads must still fail after truncate. self.insert( client, collection_name=collection_name, data=[{"id": 200, "vec": [2.0, 0.0, 0.0, 0.0], "age": 200, "tag": "old_schema"}], check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1, ct.err_msg: "Attempt to insert an unexpected field `tag`"}, ) self.query( client, collection_name=collection_name, filter="id >= 0", output_fields=["tag"], check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "field tag not exist"}, ) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) @pytest.mark.skip(reason="TODO: external collection Drop Field coverage will be validated with refresh scenarios") def test_external_collection_drop_field_rejected(self, request): """ TC-L1-20: Reject Drop Field on external collection. target: verify external collection schema cannot be mutated by Drop Field method: create an external collection, try to drop a mapped external field, then describe schema expected: Drop Field is rejected; external schema mapping remains unchanged """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() cfg = get_minio_config( minio_host=request.config.getoption("--minio_host"), minio_bucket=request.config.getoption("--minio_bucket"), ) external_source = build_external_source(cfg, f"drop-field-external/{collection_name}") # Step 1: Create an external collection with field-to-external-column mappings. schema = client.create_schema( enable_dynamic_field=False, auto_id=False, external_source=external_source, external_spec=build_external_spec(cfg), ) schema.add_field("id", DataType.INT64, external_field="id") schema.add_field("value", DataType.FLOAT, external_field="value") schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=4, external_field="embedding") client.create_collection( collection_name=collection_name, schema=schema, consistency_level="Strong", ) # Step 2: Verify the field that will be dropped is part of the external schema mapping. before_desc = client.describe_collection(collection_name) before_fields = {field["name"]: field for field in before_desc["fields"]} assert "value" in before_fields assert before_fields["value"].get("external_field") == "value" assert before_desc.get("external_source") == external_source # Step 3: Drop Field should be rejected for external collections. self.drop_collection_field( client, collection_name, field_name="value", check_task=ct.CheckTasks.err_res, check_items={ ct.err_code: 1100, ct.err_msg: "alter collection schema operation is not supported for external collection", }, ) # Step 4: Verify the failed Drop Field did not mutate schema or external field mapping. after_desc = client.describe_collection(collection_name) after_fields = {field["name"]: field for field in after_desc["fields"]} assert "value" in after_fields assert after_fields["value"].get("external_field") == "value" assert after_desc.get("external_source") == external_source assert after_desc.get("external_spec") == before_desc.get("external_spec") self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L2) def test_drop_scalar_field_type_matrix(self): """ TC-L2-21: Drop scalar field type matrix. target: verify scalar and complex fields disappear consistently after Drop Field method: create one collection with all target scalar types, then drop each field and verify output/filter/read paths expected: dropped field is invisible; explicit references fail; remaining fields work before and after reload """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() cases = [ ("flag", DataType.BOOL, {}, True, "flag == true"), ("i8", DataType.INT8, {}, 8, "i8 == 8"), ("i16", DataType.INT16, {}, 16, "i16 == 16"), ("i32", DataType.INT32, {}, 32, "i32 == 32"), ("i64", DataType.INT64, {}, 64, "i64 == 64"), ("score_f", DataType.FLOAT, {}, 1.5, "score_f > 1.0"), ("score_d", DataType.DOUBLE, {}, 2.5, "score_d > 2.0"), ("tag", DataType.VARCHAR, {"max_length": 64}, "drop_tag", 'tag == "drop_tag"'), ("payload", DataType.JSON, {}, {"k": "v", "n": 1}, 'payload["k"] == "v"'), ( "arr", DataType.ARRAY, {"element_type": DataType.INT64, "max_capacity": 8}, [1, 2, 3], "array_contains(arr, 2)", ), ("event_time", DataType.TIMESTAMPTZ, {"nullable": True}, "2025-01-02T00:00:00Z", "event_time is not null"), ] # Step 1: Create one collection that contains every scalar/complex field covered by the matrix. 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) schema.add_field("keep", DataType.INT64) for field_name, data_type, field_kwargs, _value, _filter_expr in cases: schema.add_field(field_name, data_type, **field_kwargs) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) # Step 2: Insert sealed rows where every target field has a recognizable value. rows = [] for row_id in [0, 1]: row = { "id": row_id, "vec": [float(row_id), 0.0, 0.0, 0.0], "keep": 100 + row_id, } for field_name, _data_type, _field_kwargs, value, _filter_expr in cases: row[field_name] = value rows.append(row) client.insert(collection_name=collection_name, data=rows) client.flush(collection_name) client.load_collection(collection_name) dropped_fields = [] for field_name, _data_type, _field_kwargs, _value, filter_expr in cases: # Step 3: Verify the target field is usable before it is dropped. before_query = client.query( collection_name=collection_name, filter=filter_expr, output_fields=["id", field_name], ) assert len(before_query) > 0 for row in before_query: assert field_name in row # Step 4: Drop the target scalar field and verify schema convergence. client.drop_collection_field(collection_name, field_name) dropped_fields.append(field_name) field_names = [field["name"] for field in client.describe_collection(collection_name)["fields"]] assert field_name not in field_names assert "keep" in field_names assert "vec" in field_names # Step 5: Wildcard query should not return the newly dropped field. after_query = client.query( collection_name=collection_name, filter="id in [0, 1]", output_fields=["*"], ) assert len(after_query) == 2 for row in after_query: assert "keep" in row assert field_name not in row # Step 6: Explicit output/filter references must fail with field-not-exist errors. self.query( client, collection_name=collection_name, filter="id >= 0", output_fields=[field_name], check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: f"field {field_name} not exist"}, ) self.query( client, collection_name=collection_name, filter=filter_expr, output_fields=["id"], check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: f"field {field_name} not exist"}, ) # Step 7: Search wildcard should follow the final schema after all scalar fields are dropped. after_all_drops_search = client.search( collection_name=collection_name, data=[[0.0, 0.0, 0.0, 0.0]], anns_field="vec", limit=2, output_fields=["*"], ) assert len(after_all_drops_search[0]) == 2 for hit in after_all_drops_search[0]: assert "keep" in hit["entity"] for field_name in dropped_fields: assert field_name not in hit["entity"] # Step 8: Reload and repeat the final visibility checks. client.release_collection(collection_name) client.load_collection(collection_name) after_reload = client.query( collection_name=collection_name, filter="id in [0, 1]", output_fields=["*"], ) assert len(after_reload) == 2 for row in after_reload: assert "keep" in row for field_name in dropped_fields: assert field_name not in row self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_disable_dynamic_schema_reload_reenable_and_index_cascade(self): """ TC-L1-22: Disable dynamic schema reload, re-enable, and index cascade. target: verify dynamic metadata is not resurrected after disable/reload/re-enable method: create dynamic rows and dynamic index, disable dynamic schema, reload, re-enable, then insert new dynamic rows expected: old dynamic keys/index disappear; reload does not expose old $meta; only new dynamic rows are visible after re-enable """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() dynamic_index_name = "$meta/dyn_obj/score" # Step 1: Create one dynamic-schema collection with a dynamic object used by a JSON path index. schema = client.create_schema(enable_dynamic_field=True, auto_id=False) schema.add_field("id", DataType.INT64, is_primary=True) schema.add_field("vec", DataType.FLOAT_VECTOR, dim=4) schema.add_field("age", DataType.INT64) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) # Step 2: Insert old rows with dynamic data and verify the dynamic key is visible. old_rows = [ { "id": i, "vec": [float(i), 0.0, 0.0, 0.0], "age": 20 + i, "dyn_obj": {"score": i, "tag": f"old_{i}"}, } for i in range(5) ] client.insert(collection_name=collection_name, data=old_rows) client.flush(collection_name) client.load_collection(collection_name) before_disable = client.query( collection_name=collection_name, filter='dyn_obj["score"] == 1', output_fields=["id", "dyn_obj"], ) assert len(before_disable) == 1 assert before_disable[0]["dyn_obj"]["tag"] == "old_1" # Step 3: Create a dynamic JSON path index so disable can verify index cascade cleanup. dynamic_index_params = client.prepare_index_params() dynamic_index_params.add_index( field_name="dyn_obj", index_type="INVERTED", params={"json_cast_type": "DOUBLE", "json_path": 'dyn_obj["score"]'}, ) client.create_index(collection_name, dynamic_index_params) assert dynamic_index_name in client.list_indexes(collection_name) # Step 4: Disable dynamic schema and verify dynamic metadata leaves the public schema. client.alter_collection_properties(collection_name, {"dynamicfield.enabled": False}) after_disable_desc = client.describe_collection(collection_name) assert after_disable_desc["enable_dynamic_field"] is False assert "$meta" not in [field["name"] for field in after_disable_desc["fields"]] # Step 5: Verify the dynamic index is removed by the disable operation. deadline = time.time() + 30 while dynamic_index_name in client.list_indexes(collection_name) and time.time() < deadline: time.sleep(1) assert dynamic_index_name not in client.list_indexes(collection_name) # Step 6: Reload and verify old dynamic data is still hidden from wildcard output. client.release_collection(collection_name) client.load_collection(collection_name) after_reload = client.query( collection_name=collection_name, filter="id in [0, 1, 2, 3, 4]", output_fields=["*"], ) assert len(after_reload) == 5 for row in after_reload: assert "id" in row assert "age" in row assert "vec" in row assert "dyn_obj" not in row assert "$meta" not in row # Step 7: Dynamic references must fail while dynamic schema is disabled. self.query( client, collection_name=collection_name, filter='dyn_obj["score"] == 1', output_fields=["id"], check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "field dyn_obj not exist"}, ) # Step 8: Re-enable dynamic schema and insert new rows with a new dynamic key. client.alter_collection_properties(collection_name, {"dynamicfield.enabled": True}) after_enable_desc = client.describe_collection(collection_name) assert after_enable_desc["enable_dynamic_field"] is True client.insert( collection_name=collection_name, data=[ { "id": 100, "vec": [1.0, 0.0, 0.0, 0.0], "age": 100, "dyn_after_enable": "new_value", } ], ) client.flush(collection_name) # Step 9: Verify old dynamic rows stay hidden, while new dynamic rows are visible. after_enable_rows = client.query( collection_name=collection_name, filter="id in [1, 100]", output_fields=["*"], ) rows = rows_by_id(after_enable_rows) assert "dyn_obj" not in rows[1] assert rows[100]["dyn_after_enable"] == "new_value" old_dynamic_filter = client.query( collection_name=collection_name, filter='dyn_obj["score"] == 1', output_fields=["id"], ) assert old_dynamic_filter == [] new_dynamic_filter = client.query( collection_name=collection_name, filter='dyn_after_enable == "new_value"', output_fields=["id", "dyn_after_enable"], ) assert len(new_dynamic_filter) == 1 assert new_dynamic_filter[0]["id"] == 100 assert new_dynamic_filter[0]["dyn_after_enable"] == "new_value" # Step 10: Reload once more to verify re-enabled dynamic data remains scoped to new rows only. client.release_collection(collection_name) client.load_collection(collection_name) after_final_reload = client.query( collection_name=collection_name, filter="id in [1, 100]", output_fields=["*"], ) rows = rows_by_id(after_final_reload) assert "dyn_obj" not in rows[1] assert rows[100]["dyn_after_enable"] == "new_value" self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_delete_expression_references_dropped_field(self): """ TC-L1-24: Delete expression references dropped field. target: verify delete uses the latest schema after Drop Field method: drop score, delete by kept field successfully, then reject delete by dropped score before and after reload expected: dropped-field delete fails without deleting data; kept-field delete still works """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() dynamic_collection_name = f"{collection_name}_dyn" # Step 1: Create a normal collection with score as the field to drop. 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) schema.add_field("age", DataType.INT64) schema.add_field("score", DataType.INT64) index_params = client.prepare_index_params() index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=collection_name, schema=schema, index_params=index_params, consistency_level="Strong", ) rows = [{"id": i, "vec": [float(i), 0.0, 0.0, 0.0], "age": 20 + i, "score": i * 10} for i in range(6)] client.insert(collection_name=collection_name, data=rows) client.flush(collection_name) client.load_collection(collection_name) # Step 2: Drop score and verify it leaves the schema. client.drop_collection_field(collection_name, "score") fields = [field["name"] for field in client.describe_collection(collection_name)["fields"]] assert "score" not in fields assert "age" in fields # Step 3: Delete by a kept primary-key expression should still work. delete_res = client.delete(collection_name=collection_name, filter="id in [0, 1]") assert delete_res["delete_count"] == 2 deleted_rows = client.query( collection_name=collection_name, filter="id in [0, 1]", output_fields=["id"], ) assert deleted_rows == [] remaining_before_bad_delete = client.query( collection_name=collection_name, filter="id in [2, 3, 4, 5]", output_fields=["id", "age"], ) assert {row["id"] for row in remaining_before_bad_delete} == {2, 3, 4, 5} # Step 4: Delete by dropped score must fail and must not delete any remaining rows. self.delete( client, collection_name=collection_name, filter="score > 10", check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "field score not exist"}, ) remaining_after_bad_delete = client.query( collection_name=collection_name, filter="id in [2, 3, 4, 5]", output_fields=["id", "age"], ) assert {row["id"] for row in remaining_after_bad_delete} == {2, 3, 4, 5} # Step 5: Reload and verify dropped-field delete is still rejected. client.release_collection(collection_name) client.load_collection(collection_name) self.delete( client, collection_name=collection_name, filter="score > 10", check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "field score not exist"}, ) remaining_after_reload_bad_delete = client.query( collection_name=collection_name, filter="id in [2, 3, 4, 5]", output_fields=["id", "age"], ) assert {row["id"] for row in remaining_after_reload_bad_delete} == {2, 3, 4, 5} # Step 6: Create a dynamic collection to verify same-name dynamic key is not treated as the old dropped field. dynamic_schema = client.create_schema(enable_dynamic_field=True, auto_id=False) dynamic_schema.add_field("id", DataType.INT64, is_primary=True) dynamic_schema.add_field("vec", DataType.FLOAT_VECTOR, dim=4) dynamic_schema.add_field("score", DataType.INT64) dynamic_index_params = client.prepare_index_params() dynamic_index_params.add_index(field_name="vec", index_type="AUTOINDEX", metric_type="L2") client.create_collection( collection_name=dynamic_collection_name, schema=dynamic_schema, index_params=dynamic_index_params, consistency_level="Strong", ) dynamic_rows = [ {"id": 10, "vec": [0.0, 0.0, 0.0, 0.0], "score": 10}, {"id": 11, "vec": [1.0, 0.0, 0.0, 0.0], "score": 20}, ] client.insert(collection_name=dynamic_collection_name, data=dynamic_rows) client.flush(dynamic_collection_name) client.load_collection(dynamic_collection_name) client.drop_collection_field(dynamic_collection_name, "score") # Step 7: After score is dropped, newly inserted score is dynamic data and delete(score > ...) must target only it. client.insert( collection_name=dynamic_collection_name, data=[{"id": 12, "vec": [2.0, 0.0, 0.0, 0.0], "score": 30}], ) client.flush(dynamic_collection_name) dynamic_delete_res = client.delete(dynamic_collection_name, filter="score > 25") assert dynamic_delete_res["delete_count"] == 1 old_rows = client.query( collection_name=dynamic_collection_name, filter="id in [10, 11]", output_fields=["*"], ) assert {row["id"] for row in old_rows} == {10, 11} assert all("score" not in row for row in old_rows) new_row = client.query( collection_name=dynamic_collection_name, filter="id == 12", output_fields=["id"], ) assert new_row == [] for name in [collection_name, dynamic_collection_name]: self.drop_collection(client, name)