import pytest import time from datetime import datetime, timedelta, timezone from common.common_type import CaseLabel, CheckTasks from common import common_func as cf from common import common_type as ct from utils.util_log import test_log as log from utils.util_pymilvus import * from base.client_v2_base import TestMilvusClientV2Base from pymilvus import DataType, AnnSearchRequest, WeightedRanker from pymilvus.orm.types import CONSISTENCY_STRONG, CONSISTENCY_BOUNDED, CONSISTENCY_SESSION, CONSISTENCY_EVENTUALLY default_nb = ct.default_nb default_dim = ct.default_dim default_primary_key_field_name = ct.default_primary_key_field_name default_vector_field_name = ct.default_vector_field_name default_int32_field_name = ct.default_int32_field_name default_search_exp = "id >= 0" class TestMilvusClientTTL(TestMilvusClientV2Base): """ Test case of Time To Live """ @pytest.mark.tags(CaseLabel.L0) @pytest.mark.parametrize("flush_enable", [True, False]) @pytest.mark.parametrize("on_insert", [True, False]) def test_milvus_client_ttl_default(self, flush_enable, on_insert): """ Test case for verifying TTL (Time To Live) functionality in Milvus client. This test verifies that: 1. Data becomes invisible after the specified TTL period 2. Different operations (search, query, hybrid search) correctly handle expired data 3. TTL can be altered and the changes take effect 4. Newly inserted data is not affected by previous TTL settings The test performs the following steps: 1. Create a collection with TTL enabled 2. Insert test data 3. Wait for TTL to expire and verifies data becomes invisible 4. Insert new data and verify new inserted data are visible 5. Alter TTL and verify the changes Parameters: - flush_enable: Whether to flush collection during testing - on_insert: Whether to use insert or upsert operation """ client = self._client() dim = 65 ttl = 11 nb = 1000 # field name constants pk_field = "id" vec_field = "embeddings" vec_field_2 = "embeddings_2" bool_field = "visible" collection_name = cf.gen_collection_name_by_testcase_name() schema = self.create_schema(client, enable_dynamic_field=False)[0] schema.add_field(pk_field, DataType.INT64, is_primary=True, auto_id=False) schema.add_field(vec_field, DataType.FLOAT_VECTOR, dim=dim) schema.add_field(vec_field_2, DataType.FLOAT_VECTOR, dim=dim) schema.add_field(bool_field, DataType.BOOL, nullable=True) self.create_collection(client, collection_name, schema=schema, properties={"collection.ttl.seconds": ttl}) collection_info = self.describe_collection(client, collection_name)[0] assert collection_info['properties']["collection.ttl.seconds"] == str(ttl) # create index index_params = self.prepare_index_params(client)[0] index_params.add_index(field_name=vec_field, index_type="IVF_FLAT", metric_type="COSINE", nlist=128) index_params.add_index(field_name=vec_field_2, index_type="IVF_FLAT", metric_type="COSINE", nlist=128) self.create_index(client, collection_name, index_params=index_params) # load collection self.load_collection(client, collection_name) # insert data insert_times = 2 for i in range(insert_times): start_id = i * nb rows = cf.gen_row_data_by_schema(nb=nb, schema=schema, start=start_id) for row in rows: row[bool_field] = False if on_insert is True: self.insert(client, collection_name, rows) else: self.upsert(client, collection_name, rows) # search until timeout or get empty results start_time = time.time() timeout = ttl * 5 nq = 1 search_ttl_effective = False query_ttl_effective = False hybrid_search_ttl_effective = False search_vectors = cf.gen_vectors(nq, dim=dim) sub_search1 = AnnSearchRequest(search_vectors, vec_field, {"level": 1}, 20) sub_search2 = AnnSearchRequest(search_vectors, vec_field_2, {"level": 1}, 20) ranker = WeightedRanker(0.2, 0.8) # flush collection if flush_enable is True if flush_enable: t1 = time.time() self.flush(client, collection_name) log.info(f"flush completed in {time.time() - t1}s") while time.time() - start_time < timeout: if search_ttl_effective is False: res1 = self.search(client, collection_name, search_vectors, anns_field=vec_field, search_params={"metric_type": "COSINE"}, limit=10, consistency_level=CONSISTENCY_STRONG)[0] if query_ttl_effective is False: res2 = self.query(client, collection_name, filter='', output_fields=["count(*)"], consistency_level=CONSISTENCY_STRONG)[0] if hybrid_search_ttl_effective is False: res3 = self.hybrid_search(client, collection_name, [sub_search1, sub_search2], ranker, limit=10, consistency_level=CONSISTENCY_STRONG)[0] if len(res1[0]) == 0 and search_ttl_effective is False: log.info(f"search ttl effects in {round(time.time() - start_time, 4)}s") search_ttl_effective = True if res2[0].get('count(*)', None) == 0 and query_ttl_effective is False: log.info(f"query ttl effects in {round(time.time() - start_time, 4)}s") res2x = self.query(client, collection_name, filter='visible==False', output_fields=["count(*)"], consistency_level=CONSISTENCY_STRONG)[0] log.debug(f"res2x: {res2x[0].get('count(*)', None)}") query_ttl_effective = True if len(res3[0]) == 0 and hybrid_search_ttl_effective is False: log.info(f"hybrid search ttl effects in {round(time.time() - start_time, 4)}s") hybrid_search_ttl_effective = True if search_ttl_effective is True and query_ttl_effective is True and hybrid_search_ttl_effective is True: break time.sleep(1) delta_tt = round(time.time() - start_time, 4) log.info(f"ttl effects in {delta_tt}s") assert ttl - 2 <= delta_tt <= ttl + 5 # query count(*) res = self.query(client, collection_name, filter='', output_fields=["count(*)"])[0] assert res[0].get('count(*)', None) == 0 # insert more data for i in range(insert_times): start_id = (insert_times + i) * nb rows = cf.gen_row_data_by_schema(nb=nb, schema=schema, start=start_id) for row in rows: row[bool_field] = True if on_insert is True: self.insert(client, collection_name, rows) else: self.upsert(client, collection_name, rows) # flush collection if flush_enable is True if flush_enable: t1 = time.time() self.flush(client, collection_name) log.info(f"flush completed in {time.time() - t1}s") # search data again after insert more data consistency_levels = [CONSISTENCY_EVENTUALLY, CONSISTENCY_BOUNDED, CONSISTENCY_SESSION, CONSISTENCY_STRONG] for consistency_level in consistency_levels: log.debug(f"start to search/query with {consistency_level}") # Poll until search returns results (search visibility may lag behind query) for i in range(15): res = self.search(client, collection_name, search_vectors, search_params={"metric_type": "COSINE"}, anns_field=vec_field, limit=10, consistency_level=consistency_level)[0] if len(res[0]) > 0: break time.sleep(2) assert len(res[0]) > 0, \ f"Search with {consistency_level} returned 0 results after retries" if consistency_level != CONSISTENCY_STRONG: pass else: # query count(*) res = self.query(client, collection_name, filter='', output_fields=["count(*)"], consistency_level=consistency_level)[0] assert res[0].get('count(*)', None) == nb * insert_times res = self.query(client, collection_name, filter='visible==False', output_fields=["count(*)"], consistency_level=consistency_level)[0] assert res[0].get('count(*)', None) == 0 # query count(visible) res = self.query(client, collection_name, filter='visible==True', output_fields=["count(*)"], consistency_level=consistency_level)[0] assert res[0].get('count(*)', None) == nb * insert_times # hybrid search res = self.hybrid_search(client, collection_name, [sub_search1, sub_search2], ranker, limit=10, consistency_level=consistency_level)[0] assert len(res[0]) > 0 # alter ttl to 2000s self.alter_collection_properties(client, collection_name, properties={"collection.ttl.seconds": 2000}) for consistency_level in consistency_levels: log.debug(f"start to search/query after alter ttl with {consistency_level}") # search data after alter ttl res = self.search(client, collection_name, search_vectors, search_params={"metric_type": "COSINE"}, anns_field=vec_field, filter='visible==False', limit=10, consistency_level=consistency_level, output_fields=[bool_field])[0] assert len(res[0]) > 0 for hit in res[0]: assert hit.get(bool_field) == False # hybrid search data after alter ttl sub_search1 = AnnSearchRequest(search_vectors, vec_field, {"level": 1}, 20, expr='visible==False') sub_search2 = AnnSearchRequest(search_vectors, vec_field_2, {"level": 1}, 20, expr='visible==False') res = self.hybrid_search(client, collection_name, [sub_search1, sub_search2], ranker, limit=10, consistency_level=consistency_level)[0] assert len(res[0]) > 0 # query count(*) res = self.query(client, collection_name, filter='visible==False', output_fields=["count(*)"], consistency_level=consistency_level)[0] assert res[0].get('count(*)', 0) == insert_times * nb res = self.query(client, collection_name, filter='', output_fields=["count(*)"], consistency_level=consistency_level)[0] if consistency_level != CONSISTENCY_STRONG: assert res[0].get('count(*)', 0) >= insert_times * nb else: assert res[0].get('count(*)', 0) == insert_times * nb * 2 @pytest.mark.tags(CaseLabel.L2) def test_milvus_client_ttl_edge(self): """ Test case for verifying edge case of TTL (Time To Live) functionality in Milvus client. This test verifies that: 1. Creating a collection with an extremely large TTL value should fail 2. The system should reject TTL values that are too large (e.g., 8,640,000,000,007,819,008 seconds) The test performs the following steps: 1. Attempt to create a collection with a very large TTL value 2. Verify that the creation fails with an appropriate error Expected behavior: - Collection creation should fail when TTL is set to an extremely large value - An error should be raised indicating the TTL value is invalid """ client = self._client() dim = 65 # Set an extremely large TTL value that should cause an error ttl = 9223372036854775800 collection_name = cf.gen_collection_name_by_testcase_name() schema = self.create_schema(client, enable_dynamic_field=False)[0] schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False) schema.add_field("embeddings", DataType.FLOAT_VECTOR, dim=dim) # Attempt to create collection with extremely large TTL, expecting it to fail # Use force_teardown=False since collection creation should fail error = {ct.err_code: 1100, ct.err_msg: f"collection TTL is out of range, expect [-1, 3155760000], got {ttl}: invalid parameter"} self.create_collection(client, collection_name, schema=schema, properties={"collection.ttl.seconds": ttl}, check_task=CheckTasks.err_res, check_items=error) @pytest.mark.tags(CaseLabel.L0) @pytest.mark.parametrize("partial_update", [False, True]) def test_milvus_client_partial_update_with_ttl(self, partial_update): """ target: test PU will extend the ttl of the collection method: 1. Create a collection 2. Insert rows 3. Continuously query and search the collection 4. Upsert the rows with partial update 5. query and verify ttl deadline expected: Step 5 should success """ # step 1: create collection ttl_time = 20 margin = 2 # margin zone around TTL boundaries to avoid timing races client = self._client() schema = self.create_schema(client, enable_dynamic_field=False)[0] schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False) schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim) schema.add_field(default_int32_field_name, DataType.INT32, nullable=True) index_params = self.prepare_index_params(client)[0] index_params.add_index(default_primary_key_field_name, index_type="AUTOINDEX") index_params.add_index(default_vector_field_name, index_type="AUTOINDEX") index_params.add_index(default_int32_field_name, index_type="AUTOINDEX") collection_name = cf.gen_collection_name_by_testcase_name(module_index=1) self.create_collection(client, collection_name, default_dim, schema=schema, properties={"collection.ttl.seconds": ttl_time}, consistency_level="Strong", index_params=index_params) # step 2: Insert rows rows = cf.gen_row_data_by_schema(nb=default_nb, schema=schema) self.insert(client, collection_name, rows) start_time = time.time() # start timing right after insert to align with server-side TTL calculation self.flush(client, collection_name) self.release_collection(client, collection_name) self.load_collection(client, collection_name) # step 3: Continuously query and search the collection upsert_time = ttl_time / 2 pu = True update_nb = default_nb // 2 end_time = ttl_time * 2.5 new_ttl_time = ttl_time while time.time() - start_time < end_time: # query # start_time ------- pu_time ------- ttl_time ------- new_ttl_time ------- end_time # before ttl_time, the count(*) should be default_nb # before new_ttl_time, and after ttl_time the count(*) should be update_nb # after new_ttl_time, the count(*) should be 0 elapsed = time.time() - start_time res = self.query(client, collection_name, filter=default_search_exp, output_fields=["count(*)"]) # Skip assertions near TTL boundaries to avoid timing races if elapsed < ttl_time - margin: assert res[0][0].get('count(*)') == default_nb elif elapsed > ttl_time + margin and elapsed < new_ttl_time - margin: assert res[0][0].get('count(*)') == update_nb elif elapsed > new_ttl_time + margin: assert res[0][0].get('count(*)') == 0 # search # before new_ttl_time, the search result should be 10 # after new_ttl_time, the search result should be 0 search_vectors = cf.gen_vectors(1, dim=default_dim) elapsed = time.time() - start_time res = self.search(client, collection_name, search_vectors, anns_field=default_vector_field_name, search_params={"metric_type": "COSINE"}, limit=10) if elapsed < new_ttl_time - margin: assert len(res[0][0]) == 10 elif elapsed > new_ttl_time + margin: assert len(res[0][0]) == 0 time.sleep(1) # upsert if pu and time.time() - start_time >= upsert_time: if partial_update: new_rows = cf.gen_row_data_by_schema(nb=update_nb, schema=schema, desired_field_names=[default_primary_key_field_name, default_vector_field_name]) else: new_rows = cf.gen_row_data_by_schema(nb=update_nb, schema=schema) self.upsert(client, collection_name, new_rows, partial_update=partial_update) pu_time = time.time() - start_time new_ttl_time = pu_time + ttl_time pu = False time.sleep(1) self.drop_collection(client, collection_name) class TestMilvusClientEntityTTLValid(TestMilvusClientV2Base): def _create_ttl_collection(self, client, collection_name, extra_fields=None, properties=None, ttl_nullable=True, **kwargs): """Create a collection with standard TTL schema (pk + ttl + vector + index).""" schema = self.create_schema(client, enable_dynamic_field=False)[0] schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False) schema.add_field("ttl", DataType.TIMESTAMPTZ, nullable=ttl_nullable) schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim) for field in (extra_fields or []): schema.add_field(**field) index_params = self.prepare_index_params(client)[0] index_params.add_index(field_name=default_vector_field_name, index_type="IVF_FLAT", metric_type="L2", nlist=128) if properties is None: properties = {"ttl_field": "ttl", "timezone": "UTC"} self.create_collection(client, collection_name, schema=schema, properties=properties, consistency_level="Strong", index_params=index_params, **kwargs) def _wait_until_count(self, client, collection_name, expected_count, timeout=30, interval=2): """Poll until query count(*) equals expected_count or timeout is reached.""" for _ in range(timeout // interval): res = self.query(client, collection_name, filter="", output_fields=["count(*)"], consistency_level=CONSISTENCY_STRONG)[0] if res[0].get('count(*)') == expected_count: return time.sleep(interval) res = self.query(client, collection_name, filter="", output_fields=["count(*)"], consistency_level=CONSISTENCY_STRONG)[0] assert res[0].get('count(*)') == expected_count, \ f"Expected count {expected_count}, got {res[0].get('count(*)')} after {timeout}s" def _wait_until_search_count(self, client, collection_name, search_vectors, expected_count, anns_field=default_vector_field_name, timeout=30, interval=2, **search_kwargs): """Poll until search result count equals expected_count or timeout is reached. Search and query take different code paths and TTL filtering can propagate at different speeds, so search assertions need their own retry loop — mirroring ``_wait_until_count`` for query. """ search_kwargs.setdefault("search_params", {}) search_kwargs.setdefault("limit", 10) search_kwargs.setdefault("consistency_level", CONSISTENCY_STRONG) for _ in range(timeout // interval): res = self.search(client, collection_name, search_vectors, anns_field=anns_field, **search_kwargs)[0] if len(res[0]) == expected_count: return res time.sleep(interval) res = self.search(client, collection_name, search_vectors, anns_field=anns_field, **search_kwargs)[0] assert len(res[0]) == expected_count, \ f"Expected search count {expected_count}, got {len(res[0])} after {timeout}s" return res @pytest.mark.tags(CaseLabel.L0) def test_create_collection_with_entity_ttl_field(self): """ target: test creating collection with entity ttl_field method: 1. Create a collection with ttl_field specified in properties 2. Verify ttl_field is set correctly in collection properties expected: Collection created successfully with ttl_field configured """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Create schema with Timestamptz field schema = self.create_schema(client, enable_dynamic_field=False)[0] schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False) schema.add_field("ttl", DataType.TIMESTAMPTZ, nullable=True) schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim) # Create collection with ttl_field properties = {"ttl_field": "ttl", "timezone": "UTC"} self.create_collection(client, collection_name, schema=schema, properties=properties) # Verify ttl_field is set collection_info = self.describe_collection(client, collection_name)[0] assert collection_info['properties'].get("ttl_field") == "ttl" assert collection_info['properties'].get("timezone") == "UTC" self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L0) def test_alter_add_entity_ttl_field(self): """ target: test adding ttl_field to existing collection via alter method: 1. Create a collection without ttl_field 2. Use alter_collection_properties to add ttl_field 3. Verify ttl_field is set correctly expected: ttl_field added successfully """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Create schema with Timestamptz field schema = self.create_schema(client, enable_dynamic_field=False)[0] schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False) schema.add_field("ttl", DataType.TIMESTAMPTZ, nullable=True) schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim) # Create collection without ttl_field properties = {"timezone": "UTC"} self.create_collection(client, collection_name, schema=schema, properties=properties) # Alter to add ttl_field self.alter_collection_properties(client, collection_name, properties={"ttl_field": "ttl"}) # Verify ttl_field is set collection_info = self.describe_collection(client, collection_name)[0] assert collection_info['properties'].get("ttl_field") == "ttl" self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_change_entity_ttl_field(self): """ target: test changing ttl_field to a new field method: 1. Create collection with ttl_field 2. Use alter_collection_properties to change ttl_field to a new field 3. Verify the new field is set as ttl_field expected: the new field is set as ttl_field """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Create schema schema = self.create_schema(client, enable_dynamic_field=False)[0] schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False) schema.add_field("ttl", DataType.TIMESTAMPTZ, nullable=True) schema.add_field("new_ttl", DataType.TIMESTAMPTZ, nullable=True) schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim) # Create collection with ttl_field properties = {"ttl_field": "ttl", "timezone": "UTC"} self.create_collection(client, collection_name, schema=schema, properties=properties) # Alter to change ttl_field to new_ttl self.alter_collection_properties(client, collection_name, properties={"ttl_field": "new_ttl"}) # Verify the new field is set as ttl_field collection_info = self.describe_collection(client, collection_name)[0] assert collection_info['properties'].get("ttl_field") == "new_ttl" self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_add_field_then_set_ttl_field(self): """ target: test evolving schema by adding TIMESTAMPTZ field then setting ttl_field method: 1. Create collection without TIMESTAMPTZ field 2. Use add_field to add a TIMESTAMPTZ field 3. Use alter_collection_properties to set ttl_field 4. Insert data with TTL values and verify TTL behavior works expected: Schema evolution workflow (add_field -> set ttl_field) works correctly """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 50 ttl_seconds = 8 # Create schema without TIMESTAMPTZ field schema = self.create_schema(client, enable_dynamic_field=False)[0] schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False) schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim) index_params = self.prepare_index_params(client)[0] index_params.add_index(field_name=default_vector_field_name, index_type="IVF_FLAT", metric_type="L2", nlist=128) # Create collection without ttl_field self.create_collection(client, collection_name, schema=schema, consistency_level="Strong", index_params=index_params) # Add TIMESTAMPTZ field via schema evolution self.add_collection_field(client, collection_name, field_name="ttl", data_type=DataType.TIMESTAMPTZ, nullable=True) # Reload collection after schema evolution self.release_collection(client, collection_name) self.load_collection(client, collection_name) # Set ttl_field property self.alter_collection_properties(client, collection_name, properties={"ttl_field": "ttl", "timezone": "UTC"}) # Verify ttl_field is set collection_info = self.describe_collection(client, collection_name)[0] assert collection_info['properties'].get("ttl_field") == "ttl" # Insert data with future TTL ttl_str = (datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)).isoformat() vectors = cf.gen_vectors(nb, dim=default_dim) rows = [{default_primary_key_field_name: i, "ttl": ttl_str, default_vector_field_name: list(vectors[i])} for i in range(nb)] self.insert(client, collection_name, rows) self.flush(client, collection_name) # Verify data is visible res = self.query(client, collection_name, filter="", output_fields=["count(*)"])[0] assert res[0].get('count(*)') == nb # Wait for TTL to expire (poll until count reaches 0) time.sleep(ttl_seconds) self._wait_until_count(client, collection_name, expected_count=0) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_alter_database_timezone_for_entity_ttl(self): """ target: test altering database timezone does not affect entity TTL expiration method: 1. Create database with UTC timezone 2. Create collection with ttl_field 3. Insert data with timestamptz value 4. Alter database timezone to a different timezone (e.g., Asia/Shanghai) 5. Verify TTL expiration is not affected by timezone change expected: Changing database timezone does not affect expiration of existing data with absolute timestamps """ client = self._client() db_name = cf.gen_unique_str("test_db_ttl") collection_name = cf.gen_collection_name_by_testcase_name() nb = 100 # Create database with UTC timezone self.create_database(client, db_name, properties={"timezone": "UTC"}) self.using_database(client, db_name) self._create_ttl_collection(client, collection_name, properties={"ttl_field": "ttl"}) # Insert data with future timestamp (10 seconds from now in UTC) ttl_timestamp = datetime.now(timezone.utc) + timedelta(seconds=10) ttl_str = ttl_timestamp.isoformat() vectors = cf.gen_vectors(nb, dim=default_dim) rows = [{default_primary_key_field_name: i, "ttl": ttl_str, default_vector_field_name: list(vectors[i])} for i in range(nb)] self.insert(client, collection_name, rows) self.flush(client, collection_name) # Verify data is visible res = self.query(client, collection_name, filter="", output_fields=["count(*)"])[0] assert res[0].get('count(*)') == nb # Alter database timezone to Asia/Shanghai (UTC+8) self.alter_database_properties(client, db_name, properties={"timezone": "Asia/Shanghai"}) # Verify database timezone is changed db_info = self.describe_database(client, db_name)[0] assert db_info.get("timezone") == "Asia/Shanghai" # Query again to verify data is still visible (timezone change doesn't affect existing timestamps) res = self.query(client, collection_name, filter="", output_fields=["count(*)"])[0] assert res[0].get('count(*)') == nb # Wait for TTL to expire (poll until count reaches 0) time.sleep(10) self._wait_until_count(client, collection_name, expected_count=0) # Cleanup self.drop_collection(client, collection_name) self.using_database(client, "default") self.drop_database(client, db_name) @pytest.mark.tags(CaseLabel.L0) def test_insert_with_future_entity_ttl(self): """ target: test inserting data with future ttl timestamp method: 1. Create collection with ttl_field 2. Insert data with ttl = now() + 8 seconds 3. Verify data is visible immediately 4. Wait for expiration and verify data becomes invisible expected: Data visible before TTL, invisible after TTL expires """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 100 ttl_seconds = 8 self._create_ttl_collection(client, collection_name) # Insert data with future ttl ttl_timestamp = datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds) ttl_str = ttl_timestamp.isoformat() vectors = cf.gen_vectors(nb, dim=default_dim) rows = [{default_primary_key_field_name: i, "ttl": ttl_str, default_vector_field_name: list(vectors[i])} for i in range(nb)] self.insert(client, collection_name, rows) self.flush(client, collection_name) # Verify data is visible res = self.query(client, collection_name, filter="", output_fields=["count(*)"])[0] assert res[0].get('count(*)') == nb # Wait for TTL to expire (poll until count reaches 0) time.sleep(ttl_seconds) self._wait_until_count(client, collection_name, expected_count=0) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L0) def test_insert_with_expired_entity_ttl(self): """ target: test inserting data with past ttl timestamp method: 1. Create collection with ttl_field 2. Insert data with ttl = now() - 60 seconds (already expired) 3. Query immediately to verify data is invisible expected: Data is immediately invisible after insert """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 100 self._create_ttl_collection(client, collection_name) # Insert data with past ttl (already expired) ttl_timestamp = datetime.now(timezone.utc) - timedelta(seconds=60) ttl_str = ttl_timestamp.isoformat() vectors = cf.gen_vectors(nb, dim=default_dim) rows = [{default_primary_key_field_name: i, "ttl": ttl_str, default_vector_field_name: list(vectors[i])} for i in range(nb)] self.insert(client, collection_name, rows) self.flush(client, collection_name) # Verify data is invisible (TTL filtering may take a moment to propagate) self._wait_until_count(client, collection_name, expected_count=0) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L0) def test_insert_with_null_entity_ttl(self): """ target: test inserting data with NULL ttl (never expires) persists through flush, compaction, and release/reload method: 1. Create collection with ttl_field (nullable=True) 2. Insert NULL ttl data + short TTL data 3. Wait for short TTL data to expire 4. Verify NULL data survives after flush, compact, release/reload expected: Data with NULL ttl never expires and persists through all operations """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 50 ttl_seconds = 8 self._create_ttl_collection(client, collection_name) # Insert NULL ttl data (never expires) + short TTL data future_ttl = (datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)).isoformat() vectors = cf.gen_vectors(nb * 2, dim=default_dim) rows = [] for i in range(nb * 2): ttl_value = None if i < nb else future_ttl rows.append({default_primary_key_field_name: i, "ttl": ttl_value, default_vector_field_name: list(vectors[i])}) self.insert(client, collection_name, rows) self.flush(client, collection_name) # Verify all data visible before expiry res = self.query(client, collection_name, filter="", output_fields=["count(*)"])[0] assert res[0].get('count(*)') == nb * 2 # Wait for short TTL data to expire time.sleep(ttl_seconds) self._wait_until_count(client, collection_name, expected_count=nb) # Flush and verify NULL data persists self.flush(client, collection_name) res = self.query(client, collection_name, filter="", output_fields=["count(*)"])[0] assert res[0].get('count(*)') == nb # Compact and verify NULL data persists self.compact(client, collection_name) self._wait_until_count(client, collection_name, expected_count=nb) # Release and reload, verify NULL data persists self.release_collection(client, collection_name) self.load_collection(client, collection_name) res = self.query(client, collection_name, filter="", output_fields=["count(*)"])[0] assert res[0].get('count(*)') == nb self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L0) def test_insert_mixed_entity_ttl_values(self): """ target: test inserting data with mixed ttl values (future, past, NULL) method: 1. Create collection with ttl_field 2. Insert data with: - 1/3 with ttl = now() + 8 seconds (future) - 1/3 with ttl = now() - 60 seconds (past, already expired) - 1/3 with ttl = NULL (never expires) 3. Verify only future and NULL data are visible initially 4. Wait for future data to expire 5. Verify only NULL data remains visible expected: Each data expires according to its ttl value """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 90 # 30 for each category ttl_seconds = 8 self._create_ttl_collection(client, collection_name) # Prepare ttl values future_ttl = (datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)).isoformat() past_ttl = (datetime.now(timezone.utc) - timedelta(seconds=60)).isoformat() # Insert mixed data vectors = cf.gen_vectors(nb, dim=default_dim) rows = [] for i in range(nb): if i < 30: ttl_value = future_ttl # Future elif i < 60: ttl_value = past_ttl # Past (expired) else: ttl_value = None # NULL (never expires) rows.append({default_primary_key_field_name: i, "ttl": ttl_value, default_vector_field_name: list(vectors[i])}) self.insert(client, collection_name, rows) self.flush(client, collection_name) # Verify only future + NULL data are visible (30 + 30 = 60) res = self.query(client, collection_name, filter="", output_fields=["count(*)"])[0] assert res[0].get('count(*)') == 60 # Wait for future data to expire (poll until only NULL data remains) time.sleep(ttl_seconds) self._wait_until_count(client, collection_name, expected_count=30) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L0) def test_upsert_partial_update_without_entity_ttl(self): """ target: test partial update without ttl field preserves original ttl method: 1. Create collection with ttl_field 2. Insert data with ttl = now() + 10 seconds 3. Perform partial update (only id and vector) without ttl field 4. Verify data still expires at original ttl time expected: Partial update without ttl field preserves original ttl """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 50 ttl_seconds = 10 self._create_ttl_collection(client, collection_name) # Insert data with future ttl ttl_timestamp = datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds) ttl_str = ttl_timestamp.isoformat() vectors = cf.gen_vectors(nb, dim=default_dim) rows = [{default_primary_key_field_name: i, "ttl": ttl_str, default_vector_field_name: list(vectors[i])} for i in range(nb)] self.insert(client, collection_name, rows) self.flush(client, collection_name) # Perform partial update without ttl field new_vectors = cf.gen_vectors(nb, dim=default_dim) update_rows = [{default_primary_key_field_name: i, default_vector_field_name: list(new_vectors[i])} for i in range(nb)] self.upsert(client, collection_name, update_rows, partial_update=True) # Verify data is still visible res = self.query(client, collection_name, filter="", output_fields=["count(*)"], consistency_level=CONSISTENCY_STRONG)[0] assert res[0].get('count(*)') == nb # Wait for original ttl to expire (poll until count reaches 0) time.sleep(ttl_seconds) self._wait_until_count(client, collection_name, expected_count=0) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L0) def test_upsert_extend_entity_ttl(self): """ target: test upsert with new future TTL extends entity lifetime method: 1. Create collection with ttl_field 2. Insert data with ttl = now() + 8 seconds 3. After 4 seconds, upsert same IDs with ttl = now() + 12 seconds 4. Wait past original TTL 5. Verify data is still visible (TTL was extended) 6. Wait past new TTL and verify data expires expected: Upsert with new TTL extends the expiration deadline """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 50 original_ttl_seconds = 8 self._create_ttl_collection(client, collection_name) # Insert data with original TTL (8s) original_ttl = (datetime.now(timezone.utc) + timedelta(seconds=original_ttl_seconds)).isoformat() vectors = cf.gen_vectors(nb, dim=default_dim) rows = [{default_primary_key_field_name: i, "ttl": original_ttl, default_vector_field_name: list(vectors[i])} for i in range(nb)] self.insert(client, collection_name, rows) self.flush(client, collection_name) # Wait 4 seconds, then upsert with extended TTL (12s from now) time.sleep(4) extended_ttl = (datetime.now(timezone.utc) + timedelta(seconds=12)).isoformat() upsert_rows = [{default_primary_key_field_name: i, "ttl": extended_ttl, default_vector_field_name: list(vectors[i])} for i in range(nb)] self.upsert(client, collection_name, upsert_rows) # Wait past original TTL (4 more seconds) — data should still be alive time.sleep(6) self._wait_until_count(client, collection_name, expected_count=nb) # Wait past extended TTL (poll until count reaches 0) self._wait_until_count(client, collection_name, expected_count=0) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L0) def test_upsert_shorten_entity_ttl(self): """ target: test upsert with shorter TTL makes entity expire sooner method: 1. Create collection with ttl_field 2. Insert data with ttl = now() + 60 seconds (long TTL) 3. Upsert same IDs with ttl = now() + 8 seconds (short TTL) 4. Wait for short TTL to expire 5. Verify data expires at the shorter deadline expected: Upsert with shorter TTL overrides the original deadline """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 50 short_ttl_seconds = 8 self._create_ttl_collection(client, collection_name) # Insert data with long TTL (60s) long_ttl = (datetime.now(timezone.utc) + timedelta(seconds=60)).isoformat() vectors = cf.gen_vectors(nb, dim=default_dim) rows = [{default_primary_key_field_name: i, "ttl": long_ttl, default_vector_field_name: list(vectors[i])} for i in range(nb)] self.insert(client, collection_name, rows) self.flush(client, collection_name) # Verify data is visible res = self.query(client, collection_name, filter="", output_fields=["count(*)"], consistency_level=CONSISTENCY_STRONG)[0] assert res[0].get('count(*)') == nb # Upsert with short TTL (8s from now) short_ttl = (datetime.now(timezone.utc) + timedelta(seconds=short_ttl_seconds)).isoformat() upsert_rows = [{default_primary_key_field_name: i, "ttl": short_ttl, default_vector_field_name: list(vectors[i])} for i in range(nb)] self.upsert(client, collection_name, upsert_rows) # Wait for short TTL to expire (poll until count reaches 0) time.sleep(short_ttl_seconds) self._wait_until_count(client, collection_name, expected_count=0) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_entity_ttl_after_release_reload(self): """ target: test entity TTL still works after release and reload method: 1. Create collection with ttl_field 2. Insert data with ttl = now() + 10 seconds 3. Release and reload collection 4. Verify data is still visible 5. Wait for TTL to expire 6. Verify data is invisible after TTL expires expected: TTL filtering persists across release/reload cycles """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 100 ttl_seconds = 10 self._create_ttl_collection(client, collection_name) # Insert data with future ttl ttl_timestamp = (datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)).isoformat() vectors = cf.gen_vectors(nb, dim=default_dim) rows = [{default_primary_key_field_name: i, "ttl": ttl_timestamp, default_vector_field_name: list(vectors[i])} for i in range(nb)] self.insert(client, collection_name, rows) self.flush(client, collection_name) # Verify data is visible before release res = self.query(client, collection_name, filter="", output_fields=["count(*)"], consistency_level=CONSISTENCY_STRONG)[0] assert res[0].get('count(*)') == nb # Release and reload self.release_collection(client, collection_name) self.load_collection(client, collection_name) # Verify data is still visible after reload res = self.query(client, collection_name, filter="", output_fields=["count(*)"], consistency_level=CONSISTENCY_STRONG)[0] assert res[0].get('count(*)') == nb # Wait for TTL to expire (poll until count reaches 0) time.sleep(ttl_seconds) self._wait_until_count(client, collection_name, expected_count=0) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_delete_before_entity_ttl_expires(self): """ target: test explicit delete removes entities immediately, not waiting for TTL method: 1. Create collection with ttl_field 2. Insert data with ttl = now() + 60 seconds (long TTL) 3. Delete half of the entities explicitly 4. Verify deleted entities are gone immediately 5. Verify remaining entities are still visible expected: Explicit delete takes effect immediately regardless of TTL """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 100 ttl_seconds = 60 self._create_ttl_collection(client, collection_name) # Insert data with long TTL ttl_timestamp = (datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)).isoformat() vectors = cf.gen_vectors(nb, dim=default_dim) rows = [{default_primary_key_field_name: i, "ttl": ttl_timestamp, default_vector_field_name: list(vectors[i])} for i in range(nb)] self.insert(client, collection_name, rows) self.flush(client, collection_name) # Verify all data is visible res = self.query(client, collection_name, filter="", output_fields=["count(*)"], consistency_level=CONSISTENCY_STRONG)[0] assert res[0].get('count(*)') == nb # Delete first half of entities delete_ids = list(range(nb // 2)) self.delete(client, collection_name, ids=delete_ids) # Verify deleted entities are gone immediately res = self.query(client, collection_name, filter="", output_fields=["count(*)"], consistency_level=CONSISTENCY_STRONG)[0] assert res[0].get('count(*)') == nb // 2 # Verify remaining entities are the second half res = self.query(client, collection_name, filter=f"id >= {nb // 2}", output_fields=[default_primary_key_field_name], consistency_level=CONSISTENCY_STRONG)[0] assert len(res) == nb // 2 for row in res: assert row[default_primary_key_field_name] >= nb // 2 # Verify deleted entities are not returned even by ID filter res = self.query(client, collection_name, filter=f"id < {nb // 2}", output_fields=["count(*)"], consistency_level=CONSISTENCY_STRONG)[0] assert res[0].get('count(*)') == 0 self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_insert_with_timezone_offset_entity_ttl(self): """ target: test entity TTL with explicit timezone offset timestamps method: 1. Create collection with ttl_field 2. Insert data using timestamps with explicit timezone offset (e.g., +08:00) 3. Insert data using second-level precision (no fractional seconds) 4. Verify data is visible before TTL, invisible after TTL expected: Different timestamp formats (timezone offsets, different precisions) are handled correctly """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 30 ttl_seconds = 8 self._create_ttl_collection(client, collection_name) # Insert data with explicit timezone offset (+08:00) now_utc = datetime.now(timezone.utc) future_utc = now_utc + timedelta(seconds=ttl_seconds) # Convert to +08:00 offset format tz_shanghai = timezone(timedelta(hours=8)) future_shanghai = future_utc.astimezone(tz_shanghai) ttl_with_offset = future_shanghai.strftime("%Y-%m-%dT%H:%M:%S+08:00") # Insert with second-level precision (no fractional seconds) ttl_second_precision = future_utc.strftime("%Y-%m-%dT%H:%M:%SZ") vectors = cf.gen_vectors(nb, dim=default_dim) rows = [] for i in range(nb): if i < 15: ttl_value = ttl_with_offset # Explicit timezone offset else: ttl_value = ttl_second_precision # Second-level precision with Z suffix rows.append({default_primary_key_field_name: i, "ttl": ttl_value, default_vector_field_name: list(vectors[i])}) self.insert(client, collection_name, rows) self.flush(client, collection_name) # Verify all data is visible res = self.query(client, collection_name, filter="", output_fields=["count(*)"])[0] assert res[0].get('count(*)') == nb # Wait for TTL to expire (poll until count reaches 0) time.sleep(ttl_seconds) self._wait_until_count(client, collection_name, expected_count=0) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_insert_new_data_after_entity_ttl_expiry(self): """ target: test inserting new data after previous data expired by entity TTL method: 1. Create collection with ttl_field 2. Insert first batch with ttl = now() + 8 seconds 3. Wait for TTL to expire, verify data is gone 4. Re-insert using the same PKs with new future TTL 5. Insert additional batch with new PKs 6. Verify all new data is visible and queryable expected: New inserts work normally after previous data expired. Expired PKs can be reused as open slots for new data — even though physical deletion (compaction) may not have completed, the expired data is logically invisible, and re-inserting with the same PKs succeeds via upsert semantics. """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 100 ttl_seconds = 8 self._create_ttl_collection(client, collection_name) # Insert first batch with short TTL ttl_str = (datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)).isoformat() vectors = cf.gen_vectors(nb, dim=default_dim) rows = [{default_primary_key_field_name: i, "ttl": ttl_str, default_vector_field_name: list(vectors[i])} for i in range(nb)] self.insert(client, collection_name, rows) self.flush(client, collection_name) # Verify first batch is visible res = self.query(client, collection_name, filter="", output_fields=["count(*)"], consistency_level=CONSISTENCY_STRONG)[0] assert res[0].get('count(*)') == nb # Wait for TTL to expire (poll until count reaches 0) time.sleep(ttl_seconds) self._wait_until_count(client, collection_name, expected_count=0) # Verify first batch is gone res = self.query(client, collection_name, filter="", output_fields=["count(*)"], consistency_level=CONSISTENCY_STRONG)[0] assert res[0].get('count(*)') == 0 # Re-insert using the same PKs (0..nb-1) — should succeed since originals are expired reuse_ttl_str = (datetime.now(timezone.utc) + timedelta(seconds=60)).isoformat() reuse_vectors = cf.gen_vectors(nb, dim=default_dim) reuse_rows = [{default_primary_key_field_name: i, "ttl": reuse_ttl_str, default_vector_field_name: list(reuse_vectors[i])} for i in range(nb)] self.insert(client, collection_name, reuse_rows) self.flush(client, collection_name) # Verify reused PKs are visible res = self.query(client, collection_name, filter="", output_fields=["count(*)"], consistency_level=CONSISTENCY_STRONG)[0] assert res[0].get('count(*)') == nb # Verify the reused PK data is queryable by ID res = self.query(client, collection_name, filter="id >= 0 and id < 10", output_fields=[default_primary_key_field_name, "ttl"], consistency_level=CONSISTENCY_STRONG)[0] assert len(res) == 10 # Insert additional batch with new PKs new_ttl_str = (datetime.now(timezone.utc) + timedelta(seconds=60)).isoformat() new_vectors = cf.gen_vectors(nb, dim=default_dim) new_rows = [{default_primary_key_field_name: nb + i, "ttl": new_ttl_str, default_vector_field_name: list(new_vectors[i])} for i in range(nb)] self.insert(client, collection_name, new_rows) self.flush(client, collection_name) # Verify both batches are visible (reused PKs + new PKs) res = self.query(client, collection_name, filter="", output_fields=["count(*)"], consistency_level=CONSISTENCY_STRONG)[0] assert res[0].get('count(*)') == nb * 2 # Verify search returns results from both batches search_vectors = cf.gen_vectors(1, dim=default_dim) self._wait_until_search_count(client, collection_name, search_vectors, expected_count=10) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L0) def test_query_filter_expired_entity_data(self): """ target: test query automatically filters expired data after TTL expiry method: 1. Create collection with ttl_field, insert mixed TTL data 2. Verify count before expiry (future + NULL = visible) 3. Wait for TTL to expire 4. Query and verify only NULL data remains expected: Query automatically filters expired data """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 300 ttl_seconds = 8 self._create_ttl_collection(client, collection_name) future_ttl = (datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)).isoformat() past_ttl = (datetime.now(timezone.utc) - timedelta(seconds=60)).isoformat() vectors = cf.gen_vectors(nb, dim=default_dim) rows = [] for i in range(nb): if i < 100: ttl_value = future_ttl elif i < 200: ttl_value = past_ttl else: ttl_value = None rows.append({default_primary_key_field_name: i, "ttl": ttl_value, default_vector_field_name: list(vectors[i])}) self.insert(client, collection_name, rows) self.flush(client, collection_name) # Before expiry: future (100) + NULL (100) = 200 visible; past (100) already expired res = self.query(client, collection_name, filter="", output_fields=["count(*)"])[0] assert res[0].get('count(*)') == 200 # Wait for TTL to expire (poll until only NULL data remains) time.sleep(ttl_seconds) self._wait_until_count(client, collection_name, expected_count=100) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L0) def test_search_filter_expired_entity_data(self): """ target: test search filters expired data when filter expression is present method: 1. Create collection with ttl_field, insert future TTL + NULL TTL data 2. Wait for TTL to expire 3. Search with filter and verify only NULL TTL data is returned expected: Search results do not include expired data """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 200 ttl_seconds = 8 self._create_ttl_collection(client, collection_name) future_ttl = (datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)).isoformat() vectors = cf.gen_vectors(nb, dim=default_dim) rows = [] for i in range(nb): if i < 100: ttl_value = future_ttl else: ttl_value = None rows.append({default_primary_key_field_name: i, "ttl": ttl_value, default_vector_field_name: list(vectors[i])}) self.insert(client, collection_name, rows) self.flush(client, collection_name) # Wait for TTL to expire (poll until only NULL data remains) time.sleep(ttl_seconds) self._wait_until_count(client, collection_name, expected_count=100) # Poll until search only returns non-expired entities. # Search TTL filtering can lag behind query, so retry. search_vectors = cf.gen_vectors(1, dim=default_dim) for _ in range(15): res = self.search(client, collection_name, search_vectors, anns_field=default_vector_field_name, search_params={}, limit=10, filter="id >= 0", consistency_level=CONSISTENCY_STRONG)[0] if len(res[0]) > 0 and all(hit['id'] >= 100 for hit in res[0]): break time.sleep(2) assert len(res[0]) > 0 for hit in res[0]: assert hit['id'] >= 100 self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_search_without_filter_expired_entity_data(self): """ target: test search without filter also filters expired data method: 1. Create collection with ttl_field, insert future TTL + NULL TTL data 2. Wait for TTL to expire 3. Search without filter and verify only NULL TTL data is returned expected: TTL filtering is applied regardless of whether a filter expression is present. Search without filter should not return expired entities. """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 200 ttl_seconds = 8 self._create_ttl_collection(client, collection_name) future_ttl = (datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)).isoformat() vectors = cf.gen_vectors(nb, dim=default_dim) rows = [] for i in range(nb): if i < 100: ttl_value = future_ttl else: ttl_value = None rows.append({default_primary_key_field_name: i, "ttl": ttl_value, default_vector_field_name: list(vectors[i])}) self.insert(client, collection_name, rows) self.flush(client, collection_name) # Wait for TTL to expire (poll until only NULL data remains) time.sleep(ttl_seconds) self._wait_until_count(client, collection_name, expected_count=100) # Poll until search only returns non-expired (NULL TTL) entities. # Search TTL filtering can lag behind query, so retry. search_vectors = cf.gen_vectors(1, dim=default_dim) for _ in range(15): res = self.search(client, collection_name, search_vectors, anns_field=default_vector_field_name, search_params={}, limit=10, consistency_level=CONSISTENCY_STRONG)[0] if len(res[0]) > 0 and all(hit['id'] >= 100 for hit in res[0]): break time.sleep(2) assert len(res[0]) > 0 for hit in res[0]: assert hit['id'] >= 100, \ f"Search without filter returned expired entity id={hit['id']} (expected only id >= 100)" self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L0) def test_query_output_entity_ttl_field(self): """ target: test query can output ttl field values method: 1. Create collection with ttl_field, insert data with NULL ttl 2. Query with output_fields including ttl 3. Verify ttl values are returned correctly expected: ttl field values are returned in query results """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 10 self._create_ttl_collection(client, collection_name, extra_fields=[ {"field_name": "varchar_field", "datatype": DataType.VARCHAR, "max_length": 100, "nullable": True} ]) vectors = cf.gen_vectors(nb, dim=default_dim) rows = [{default_primary_key_field_name: i, "ttl": None, default_vector_field_name: list(vectors[i]), "varchar_field": f"text_{i}"} for i in range(nb)] self.insert(client, collection_name, rows) self.flush(client, collection_name) res = self.query(client, collection_name, filter=f"id >= 0 and id < 5", output_fields=[default_primary_key_field_name, "ttl", "varchar_field"], consistency_level=CONSISTENCY_STRONG)[0] assert len(res) == 5 for row in res: assert default_primary_key_field_name in row assert "ttl" in row assert row["ttl"] is None self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_entity_ttl_compaction_reclaims_space(self): """ target: test that compaction physically removes expired TTL data and reclaims space method: 1. Create collection with ttl_field, insert data with short TTL 2. Record row_count from collection stats before expiry 3. Wait for TTL to expire, then trigger compaction 4. Verify row_count decreases after compaction completes expected: Compaction physically deletes expired data and reduces row_count """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 200 ttl_seconds = 8 self._create_ttl_collection(client, collection_name) # Insert data with short TTL future_ttl = (datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)).isoformat() vectors = cf.gen_vectors(nb, dim=default_dim) rows = [{default_primary_key_field_name: i, "ttl": future_ttl, default_vector_field_name: list(vectors[i])} for i in range(nb)] self.insert(client, collection_name, rows) self.flush(client, collection_name) # Record row_count before expiry stats_before = self.get_collection_stats(client, collection_name)[0] row_count_before = stats_before.get("row_count", 0) log.info(f"Row count before expiry: {row_count_before}") # Wait for TTL to expire time.sleep(ttl_seconds) self._wait_until_count(client, collection_name, expected_count=0) # Trigger compaction and wait for it to complete self.compact(client, collection_name) time.sleep(10) # Verify row_count decreases after compaction stats_after = self.get_collection_stats(client, collection_name)[0] row_count_after = stats_after.get("row_count", 0) log.info(f"Row count after compaction: {row_count_after}") assert row_count_after < row_count_before, \ f"Expected row_count to decrease after compaction, before={row_count_before}, after={row_count_after}" self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_entity_ttl_with_partitions(self): """ target: test entity TTL works independently across different partitions method: 1. Create collection with ttl_field and two partitions 2. Insert short TTL data into partition_a, long TTL data into partition_b 3. Wait for short TTL to expire 4. Verify partition_a data expired, partition_b data still visible expected: TTL expiration is per-entity and works correctly across partitions """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 50 short_ttl_seconds = 8 self._create_ttl_collection(client, collection_name) self.create_partition(client, collection_name, "partition_a") self.create_partition(client, collection_name, "partition_b") # Insert short TTL data into partition_a short_ttl = (datetime.now(timezone.utc) + timedelta(seconds=short_ttl_seconds)).isoformat() vectors_a = cf.gen_vectors(nb, dim=default_dim) rows_a = [{default_primary_key_field_name: i, "ttl": short_ttl, default_vector_field_name: list(vectors_a[i])} for i in range(nb)] self.insert(client, collection_name, rows_a, partition_name="partition_a") # Insert long TTL data into partition_b long_ttl = (datetime.now(timezone.utc) + timedelta(seconds=300)).isoformat() vectors_b = cf.gen_vectors(nb, dim=default_dim) rows_b = [{default_primary_key_field_name: nb + i, "ttl": long_ttl, default_vector_field_name: list(vectors_b[i])} for i in range(nb)] self.insert(client, collection_name, rows_b, partition_name="partition_b") self.flush(client, collection_name) # Verify both partitions have data res = self.query(client, collection_name, filter="", output_fields=["count(*)"], partition_names=["partition_a"])[0] assert res[0].get('count(*)') == nb res = self.query(client, collection_name, filter="", output_fields=["count(*)"], partition_names=["partition_b"])[0] assert res[0].get('count(*)') == nb # Wait for short TTL to expire time.sleep(short_ttl_seconds) # Poll on total count: partition_a expired (0) + partition_b alive (nb) = nb self._wait_until_count(client, collection_name, expected_count=nb) # Verify partition_a data expired (poll in case per-partition propagation lags) for _ in range(15): res = self.query(client, collection_name, filter="", output_fields=["count(*)"], partition_names=["partition_a"], consistency_level=CONSISTENCY_STRONG)[0] if res[0].get('count(*)') == 0: break time.sleep(2) assert res[0].get('count(*)') == 0, \ f"Expected partition_a count 0, got {res[0].get('count(*)')}" # Verify partition_b data still visible res = self.query(client, collection_name, filter="", output_fields=["count(*)"], partition_names=["partition_b"], consistency_level=CONSISTENCY_STRONG)[0] assert res[0].get('count(*)') == nb self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_entity_ttl_switch_ttl_field(self): """ target: test switching ttl_field from one TIMESTAMPTZ field to another method: 1. Create collection with two TIMESTAMPTZ fields (ttl, ttl_dynamic), set ttl as ttl_field 2. Insert data where ttl = far future (won't expire), ttl_dynamic = short TTL 3. Verify data does NOT expire based on ttl_dynamic (not the active ttl_field) 4. Switch ttl_field to ttl_dynamic via alter_collection_properties 5. Verify data now expires based on ttl_dynamic expected: Expiration behavior follows the currently active ttl_field; switching ttl_field changes which field controls expiry """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 50 short_ttl_seconds = 8 # Create schema with two TIMESTAMPTZ fields self._create_ttl_collection(client, collection_name, extra_fields=[ {"field_name": "ttl_dynamic", "datatype": DataType.TIMESTAMPTZ, "nullable": True} ]) # ttl = far future (won't expire under original ttl_field) # ttl_dynamic = short TTL (will expire if switched to) far_future = (datetime.now(timezone.utc) + timedelta(seconds=300)).isoformat() short_ttl = (datetime.now(timezone.utc) + timedelta(seconds=short_ttl_seconds)).isoformat() vectors = cf.gen_vectors(nb, dim=default_dim) rows = [{default_primary_key_field_name: i, "ttl": far_future, "ttl_dynamic": short_ttl, default_vector_field_name: list(vectors[i])} for i in range(nb)] self.insert(client, collection_name, rows) self.flush(client, collection_name) # Verify all data is visible (ttl_field=ttl, ttl is far future) res = self.query(client, collection_name, filter="", output_fields=["count(*)"])[0] assert res[0].get('count(*)') == nb # Wait past short_ttl_seconds — data should still be alive because # the active ttl_field is "ttl" (far future), not "ttl_dynamic" time.sleep(short_ttl_seconds + 3) res = self.query(client, collection_name, filter="", output_fields=["count(*)"])[0] assert res[0].get('count(*)') == nb, \ "Data should NOT expire based on inactive ttl_dynamic field" # Switch ttl_field to ttl_dynamic self.alter_collection_properties(client, collection_name, properties={"ttl_field": "ttl_dynamic"}) # Verify ttl_field is updated collection_info = self.describe_collection(client, collection_name)[0] assert collection_info['properties'].get("ttl_field") == "ttl_dynamic" # ttl_dynamic timestamps are already past — data should become invisible # after switching (property change may take a moment to propagate) self._wait_until_count(client, collection_name, expected_count=0) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_entity_ttl_with_query_iterator(self): """ target: test query iterator filters expired TTL data method: 1. Create collection with ttl_field 2. Insert short TTL data + NULL TTL data 3. Wait for short TTL to expire 4. Use query_iterator to traverse all data 5. Verify iterator only returns non-expired (NULL TTL) data expected: Query iterator respects TTL filtering, expired entities are not yielded """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 50 ttl_seconds = 8 self._create_ttl_collection(client, collection_name) # Insert short TTL data (id 0~49) + NULL TTL data (id 50~99) future_ttl = (datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)).isoformat() vectors = cf.gen_vectors(nb * 2, dim=default_dim) rows = [] for i in range(nb * 2): ttl_value = future_ttl if i < nb else None rows.append({default_primary_key_field_name: i, "ttl": ttl_value, default_vector_field_name: list(vectors[i])}) self.insert(client, collection_name, rows) self.flush(client, collection_name) # Wait for short TTL to expire time.sleep(ttl_seconds) self._wait_until_count(client, collection_name, expected_count=nb) # Use query_iterator to traverse all remaining data iterator = self.query_iterator(client, collection_name, batch_size=10, output_fields=[default_primary_key_field_name, "ttl"])[0] iterated_ids = [] while True: batch = iterator.next() if not batch: break for row in batch: iterated_ids.append(row[default_primary_key_field_name]) iterator.close() # Verify iterator only returned NULL TTL data (id >= 50) assert len(iterated_ids) == nb, \ f"Expected {nb} rows from iterator, got {len(iterated_ids)}" for pk in iterated_ids: assert pk >= nb, \ f"Iterator returned expired entity id={pk} (expected only id >= {nb})" self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L1) def test_upsert_overwrite_null_ttl_with_short_ttl(self): """ target: test that upserting the same PK with a short TTL overwrites the original NULL (never-expire) TTL, and the data expires correctly across growing segments, sealed segments, release/reload, and compaction method: 1. Create collection with ttl_field 2. Insert data with NULL ttl (never expires) and flush to seal the segment 3. Upsert same PKs with ttl = now() + 5 seconds (short TTL) 4. Wait for TTL to expire, then query and search to verify data is gone 5. Release and reload — verify expired data (including original NULL rows) remains invisible 6. Trigger compaction and verify data is still invisible expected: Upsert overwrites the original NULL TTL; after expiry the data is invisible through query, search, release/reload, and compaction """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() nb = 100 ttl_seconds = 5 self._create_ttl_collection(client, collection_name) # Step 1: Insert data with NULL ttl (never expires) and flush to seal vectors = cf.gen_vectors(nb, dim=default_dim) rows = [{default_primary_key_field_name: i, "ttl": None, default_vector_field_name: list(vectors[i])} for i in range(nb)] self.insert(client, collection_name, rows) self.flush(client, collection_name) # Verify all data is visible res = self.query(client, collection_name, filter="", output_fields=["count(*)"], consistency_level=CONSISTENCY_STRONG)[0] assert res[0].get('count(*)') == nb # Step 2: Upsert same PKs with short TTL short_ttl = (datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)).isoformat() upsert_rows = [{default_primary_key_field_name: i, "ttl": short_ttl, default_vector_field_name: list(vectors[i])} for i in range(nb)] self.upsert(client, collection_name, upsert_rows) # Verify data is still visible before expiry res = self.query(client, collection_name, filter="", output_fields=["count(*)"], consistency_level=CONSISTENCY_STRONG)[0] assert res[0].get('count(*)') == nb # Step 3: Wait for TTL to expire, verify via query and search time.sleep(ttl_seconds) self._wait_until_count(client, collection_name, expected_count=0) search_vectors = cf.gen_vectors(1, dim=default_dim) self._wait_until_search_count(client, collection_name, search_vectors, expected_count=0) # Step 4: Release and reload — expired data and original NULL rows must stay invisible self.release_collection(client, collection_name) self.load_collection(client, collection_name) res = self.query(client, collection_name, filter="", output_fields=["count(*)"], consistency_level=CONSISTENCY_STRONG)[0] assert res[0].get('count(*)') == 0, \ f"Expected 0 after release/reload, got {res[0].get('count(*)')}" self._wait_until_search_count(client, collection_name, search_vectors, expected_count=0) # Step 5: Compact and verify data is still invisible self.compact(client, collection_name) time.sleep(10) self._wait_until_count(client, collection_name, expected_count=0) self._wait_until_search_count(client, collection_name, search_vectors, expected_count=0) self.drop_collection(client, collection_name) class TestMilvusClientEntityTTLInvalid(TestMilvusClientV2Base): @pytest.mark.tags(CaseLabel.L1) def test_entity_ttl_field_nullable_false(self): """ target: test inserting NULL to non-nullable ttl field should fail method: 1. Create collection with ttl_field (nullable=False) 2. Attempt to insert data with ttl = NULL 3. Verify insert fails expected: Insert fails when ttl is NULL and field is not nullable """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Create schema with non-nullable ttl field schema = self.create_schema(client, enable_dynamic_field=False)[0] schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False) schema.add_field("ttl", DataType.TIMESTAMPTZ, nullable=False) # Not nullable schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim) # Create collection with ttl_field properties = {"ttl_field": "ttl", "timezone": "UTC"} self.create_collection(client, collection_name, schema=schema, properties=properties) # Create index and load index_params = self.prepare_index_params(client)[0] index_params.add_index(field_name=default_vector_field_name, index_type="IVF_FLAT", metric_type="L2", nlist=128) self.create_index(client, collection_name, index_params=index_params) self.load_collection(client, collection_name) # Attempt to insert NULL ttl - should fail vectors = cf.gen_vectors(10, dim=default_dim) rows = [{default_primary_key_field_name: i, "ttl": None, default_vector_field_name: list(vectors[i])} for i in range(10)] error = {ct.err_code: 1100, ct.err_msg: "the num_rows (0) of field (ttl) is not equal to passed num_rows (10): invalid parameter[expected=10][actual=0]"} self.insert(client, collection_name, rows, check_task=CheckTasks.err_res, check_items=error) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.L2) def test_create_collection_entity_ttl_field_not_timestamptz(self): """ target: test creating collection with non-Timestamptz ttl_field should fail method: 1. Create schema with Int64 field 2. Attempt to create collection with ttl_field pointing to Int64 field 3. Verify creation fails expected: Collection creation fails when ttl_field is not Timestamptz type """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Create schema with Int64 field (not Timestamptz) schema = self.create_schema(client, enable_dynamic_field=False)[0] schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False) schema.add_field("int_field", DataType.INT64, nullable=True) # Not Timestamptz schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim) # Attempt to create collection with int_field as ttl_field - should fail properties = {"ttl_field": "int_field", "timezone": "UTC"} error = {ct.err_code: 1100, ct.err_msg: "ttl field must be timestamptz, field name = int_field: invalid parameter"} self.create_collection(client, collection_name, schema=schema, properties=properties, check_task=CheckTasks.err_res, check_items=error) @pytest.mark.tags(CaseLabel.L2) def test_create_collection_entity_ttl_field_nonexistent(self): """ target: test creating collection with non-existent ttl_field should fail method: 1. Create schema without the specified ttl field 2. Attempt to create collection with ttl_field pointing to non-existent field 3. Verify creation fails expected: Collection creation fails when ttl_field does not exist """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Create schema without ttl field schema = self.create_schema(client, enable_dynamic_field=False)[0] schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False) schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim) # Attempt to create collection with non-existent ttl_field - should fail properties = {"ttl_field": "nonexistent_field", "timezone": "UTC"} error = {ct.err_code: 1100, ct.err_msg: "ttl field name nonexistent_field not found in schema: invalid parameter"} self.create_collection(client, collection_name, schema=schema, properties=properties, check_task=CheckTasks.err_res, check_items=error) @pytest.mark.tags(CaseLabel.L2) def test_create_collection_both_entity_and_collection_ttl(self): """ target: test creating collection with both ttl_field and collection.ttl.seconds should fail method: 1. Create schema with Timestamptz field 2. Attempt to create collection with both ttl_field and collection.ttl.seconds 3. Verify creation fails with mutual exclusion error expected: Collection creation fails when both TTL types are specified """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Create schema schema = self.create_schema(client, enable_dynamic_field=False)[0] schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False) schema.add_field("ttl", DataType.TIMESTAMPTZ, nullable=True) schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim) # Attempt to create collection with both TTL configs - should fail properties = { "ttl_field": "ttl", "collection.ttl.seconds": 3600, "timezone": "UTC" } error = {ct.err_code: 1100, ct.err_msg: "collection TTL and ttl field cannot be set at the same time: invalid parameter"} self.create_collection(client, collection_name, schema=schema, properties=properties, check_task=CheckTasks.err_res, check_items=error) @pytest.mark.tags(CaseLabel.L2) def test_alter_add_entity_ttl_with_existing_collection_ttl(self): """ target: test adding ttl_field when collection.ttl.seconds already exists should fail method: 1. Create collection with collection.ttl.seconds 2. Attempt to add ttl_field via alter 3. Verify alter fails expected: Cannot add ttl_field when collection.ttl.seconds is set """ client = self._client() collection_name = cf.gen_collection_name_by_testcase_name() # Create schema schema = self.create_schema(client, enable_dynamic_field=False)[0] schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False) schema.add_field("ttl", DataType.TIMESTAMPTZ, nullable=True) schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim) # Create collection with collection.ttl.seconds properties = {"collection.ttl.seconds": 3600} self.create_collection(client, collection_name, schema=schema, properties=properties) # Attempt to add ttl_field - should fail error = {ct.err_code: 1100, ct.err_msg: "collection TTL is already set, cannot be set ttl field: invalid parameter"} self.alter_collection_properties(client, collection_name, properties={"ttl_field": "ttl"}, check_task=CheckTasks.err_res, check_items=error) self.drop_collection(client, collection_name)