wehub-resource-sync
498b235461
Build and test / Build and test AMD64 Ubuntu 22.04 (push) Failing after 0s
Publish Builder / amazonlinux2023 (push) Failing after 1s
Build and test / UT for Go (push) Has been skipped
Publish KRTE Images / KRTE (push) Failing after 1s
Build and test / Integration Test (push) Has been skipped
Build and test / Upload Code Coverage (push) Has been skipped
Publish Builder / rockylinux9 (push) Failing after 1s
Publish Builder / ubuntu22.04 (push) Failing after 0s
Publish Builder / ubuntu24.04 (push) Failing after 0s
Publish Gpu Builder / publish-gpu-builder (push) Failing after 1s
Publish Test Images / PyTest (push) Failing after 0s
Build and test / UT for Cpp (push) Has been cancelled
412 lines
20 KiB
Python
412 lines
20 KiB
Python
import math
|
||
import pytest
|
||
import time
|
||
from check import param_check as pc
|
||
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
|
||
|
||
# Test parameters
|
||
default_nb = ct.default_nb
|
||
default_limit = ct.default_limit
|
||
default_search_exp = "id >= 0"
|
||
|
||
|
||
class TestMilvusClientE2E(TestMilvusClientV2Base):
|
||
""" Test case of end-to-end interface """
|
||
|
||
@pytest.mark.tags(CaseLabel.L0)
|
||
@pytest.mark.parametrize("flush_enable", [True, False])
|
||
@pytest.mark.parametrize("scalar_index_enable", [True, False])
|
||
def test_milvus_client_e2e_default(self, flush_enable, scalar_index_enable):
|
||
"""
|
||
target: test full E2E lifecycle with all nullable scalar types and nullable vector
|
||
method: 1. create collection with nullable fields (bool, int8/16/32/64, float, double, varchar, json, array, vector)
|
||
2. insert 6000 rows (2 batches × 3000) with ~20% nulls
|
||
3. create vector index + optional scalar indexes
|
||
4. search with COSINE metric, verify distance ordering and no NaN (nullable vector)
|
||
5. query with filters on each scalar type: null/not-null/comparison/range/like/in
|
||
6. delete all data, verify search and query return empty
|
||
expected: all search/query results match locally computed expected data;
|
||
no NaN distances from nullable vector; deletion fully effective
|
||
"""
|
||
client = self._client()
|
||
dim = 8
|
||
vector_type = DataType.FLOAT_VECTOR
|
||
|
||
# 1. Create collection with custom schema
|
||
collection_name = cf.gen_collection_name_by_testcase_name()
|
||
schema = self.create_schema(client, enable_dynamic_field=False)[0]
|
||
# Primary key and vector field
|
||
schema.add_field("id", DataType.INT64, is_primary=True, auto_id=False)
|
||
schema.add_field("vector", vector_type, dim=dim, nullable=True)
|
||
# Boolean type
|
||
schema.add_field("bool_field", DataType.BOOL, nullable=True)
|
||
# Integer types
|
||
schema.add_field("int8_field", DataType.INT8, nullable=True)
|
||
schema.add_field("int16_field", DataType.INT16, nullable=True)
|
||
schema.add_field("int32_field", DataType.INT32, nullable=True)
|
||
schema.add_field("int64_field", DataType.INT64, nullable=True)
|
||
# Float types
|
||
schema.add_field("float_field", DataType.FLOAT, nullable=True)
|
||
schema.add_field("double_field", DataType.DOUBLE, nullable=True)
|
||
# String type
|
||
schema.add_field("varchar_field", DataType.VARCHAR, max_length=65535, nullable=True)
|
||
# JSON type
|
||
schema.add_field("json_field", DataType.JSON, nullable=True)
|
||
# Array type
|
||
schema.add_field("array_field", DataType.ARRAY, element_type=DataType.FLOAT, max_capacity=12, nullable=True)
|
||
|
||
# Create collection
|
||
self.create_collection(client, collection_name, schema=schema)
|
||
|
||
# 2. Insert data with null values for nullable fields
|
||
num_inserts = 2 # 2 batches to cover sealed + growing scenarios
|
||
total_rows = []
|
||
for i in range(num_inserts):
|
||
data = cf.gen_row_data_by_schema(nb=default_nb, schema=schema, start=i * default_nb)
|
||
self.insert(client, collection_name, data)
|
||
total_rows.extend(data)
|
||
log.info(f"Total inserted {num_inserts * default_nb} entities")
|
||
|
||
if flush_enable:
|
||
self.flush(client, collection_name)
|
||
log.info("Flush enabled: executing flush operation")
|
||
|
||
# Create index parameters
|
||
index_params = self.prepare_index_params(client)[0]
|
||
index_params.add_index("vector", metric_type="COSINE")
|
||
|
||
# Add autoindex for scalar fields if enabled
|
||
if scalar_index_enable:
|
||
index_params.add_index(field_name="int8_field", index_type="AUTOINDEX")
|
||
index_params.add_index(field_name="int16_field", index_type="AUTOINDEX")
|
||
index_params.add_index(field_name="int32_field", index_type="AUTOINDEX")
|
||
index_params.add_index(field_name="int64_field", index_type="AUTOINDEX")
|
||
index_params.add_index(field_name="float_field", index_type="AUTOINDEX")
|
||
index_params.add_index(field_name="double_field", index_type="AUTOINDEX")
|
||
index_params.add_index(field_name="varchar_field", index_type="AUTOINDEX")
|
||
index_params.add_index(field_name="array_field", index_type="AUTOINDEX")
|
||
|
||
# 3. create index
|
||
self.create_index(client, collection_name, index_params)
|
||
|
||
# Verify scalar indexes are created if enabled
|
||
indexes = self.list_indexes(client, collection_name)[0]
|
||
log.info(f"Created indexes: {indexes}")
|
||
expected_scalar_indexes = ["int8_field", "int16_field", "int32_field", "int64_field",
|
||
"float_field", "double_field", "varchar_field", "array_field"]
|
||
if scalar_index_enable:
|
||
for field in expected_scalar_indexes:
|
||
assert field in indexes, f"Scalar index not created for field: {field}"
|
||
else:
|
||
for field in expected_scalar_indexes:
|
||
assert field not in indexes, f"Scalar index should not be created for field: {field}"
|
||
|
||
# 4. Load collection
|
||
t0 = time.time()
|
||
self.load_collection(client, collection_name)
|
||
t1 = time.time()
|
||
log.info(f"Load collection cost {t1 - t0:.4f} seconds")
|
||
|
||
# 5. Search
|
||
t0 = time.time()
|
||
vectors_to_search = cf.gen_vectors(1, dim, vector_data_type=vector_type)
|
||
search_params = {"metric_type": "COSINE", "params": {"nprobe": 100}}
|
||
search_res, _ = self.search(
|
||
client,
|
||
collection_name,
|
||
vectors_to_search,
|
||
anns_field="vector",
|
||
search_params=search_params,
|
||
limit=default_limit,
|
||
output_fields=['*'],
|
||
check_task=CheckTasks.check_search_results,
|
||
check_items={"enable_milvus_client_api": True,
|
||
"nq": len(vectors_to_search),
|
||
"pk_name": "id",
|
||
"limit": default_limit,
|
||
"metric": "COSINE"}
|
||
)
|
||
# Verify no NaN distances (nullable vector leak detection)
|
||
for hits in search_res:
|
||
for hit in hits:
|
||
assert not math.isnan(hit["distance"]), \
|
||
f"NaN distance found in search result, pk={hit['id']}"
|
||
t1 = time.time()
|
||
log.info(f"Search cost {t1 - t0:.4f} seconds")
|
||
|
||
# 6. Query with filters on each scalar field
|
||
t0 = time.time()
|
||
# Data-driven query cases: (filter_string, predicate_lambda, with_vec, description)
|
||
query_cases = [
|
||
# Boolean field (with_vec=False: skip nullable vector comparison in check)
|
||
("bool_field == true",
|
||
lambda r: r["bool_field"] is not None and r["bool_field"] is True,
|
||
False, "bool true"),
|
||
# Int8: null or < 10
|
||
("int8_field is null || int8_field < 10",
|
||
lambda r: r["int8_field"] is None or r["int8_field"] < 10,
|
||
True, "int8 null or < 10"),
|
||
# Int16: range [100, 200)
|
||
("100 <= int16_field < 200",
|
||
lambda r: r["int16_field"] is not None and 100 <= r["int16_field"] < 200,
|
||
True, "int16 range [100, 200)"),
|
||
# Int32: in set
|
||
("int32_field in [1,2,5,6]",
|
||
lambda r: r["int32_field"] is not None and r["int32_field"] in [1, 2, 5, 6],
|
||
True, "int32 in [1,2,5,6]"),
|
||
# Int64: range [4678, 5050)
|
||
("int64_field >= 4678 and int64_field < 5050",
|
||
lambda r: r["int64_field"] is not None and r["int64_field"] >= 4678 and r["int64_field"] < 5050,
|
||
True, "int64 range [4678, 5050)"),
|
||
# Float: (0.5, 0.7]
|
||
("float_field > 0.5 and float_field <= 0.7",
|
||
lambda r: r["float_field"] is not None and r["float_field"] > 0.5 and r["float_field"] <= 0.7,
|
||
True, "float (0.5, 0.7]"),
|
||
# Double: [0.5, 0.7]
|
||
("0.5 <=double_field <= 0.7",
|
||
lambda r: r["double_field"] is not None and 0.5 <= r["double_field"] <= 0.7,
|
||
True, "double [0.5, 0.7]"),
|
||
# Varchar: like prefix
|
||
('varchar_field like "varchar_1%"',
|
||
lambda r: r["varchar_field"] is not None and r["varchar_field"].startswith("varchar_1"),
|
||
True, "varchar like varchar_1%"),
|
||
# Varchar: is null
|
||
("varchar_field is null",
|
||
lambda r: r["varchar_field"] is None,
|
||
True, "varchar is null"),
|
||
# JSON: is null
|
||
("json_field is null",
|
||
lambda r: r["json_field"] is None,
|
||
True, "json is null"),
|
||
# Array: is null
|
||
("array_field is null",
|
||
lambda r: r["array_field"] is None,
|
||
True, "array is null"),
|
||
# Multiple fields all null
|
||
("varchar_field is null and json_field is null and array_field is null",
|
||
lambda r: r["varchar_field"] is None and r["json_field"] is None and r["array_field"] is None,
|
||
True, "multi fields all null"),
|
||
# Mix: varchar null and json not null
|
||
("varchar_field is null and json_field is not null",
|
||
lambda r: r["varchar_field"] is None and r["json_field"] is not None,
|
||
True, "varchar null and json not null"),
|
||
# Int8: not null and > 100
|
||
("int8_field is not null and int8_field > 100",
|
||
lambda r: r["int8_field"] is not None and r["int8_field"] > 100,
|
||
True, "int8 not null and > 100"),
|
||
# Int16: not null and < 100
|
||
("int16_field is not null and int16_field < 100",
|
||
lambda r: r["int16_field"] is not None and r["int16_field"] < 100,
|
||
True, "int16 not null and < 100"),
|
||
# Float: not null and (0.5, 0.7]
|
||
("float_field is not null and float_field > 0.5 and float_field <= 0.7",
|
||
lambda r: r["float_field"] is not None and r["float_field"] > 0.5 and r["float_field"] <= 0.7,
|
||
True, "float not null and (0.5, 0.7]"),
|
||
# Double: not null and <= 0.2
|
||
("double_field is not null and double_field <= 0.2",
|
||
lambda r: r["double_field"] is not None and r["double_field"] <= 0.2,
|
||
True, "double not null and <= 0.2"),
|
||
# Varchar: not null
|
||
("varchar_field is not null",
|
||
lambda r: r["varchar_field"] is not None,
|
||
True, "varchar not null"),
|
||
# JSON: not null and count < 15
|
||
("json_field is not null and json_field['count'] < 15",
|
||
lambda r: r["json_field"] is not None and r["json_field"]["count"] < 15,
|
||
True, "json not null and count < 15"),
|
||
# Array: not null and first element < 100
|
||
("array_field is not null and array_field[0] < 100",
|
||
lambda r: r["array_field"] is not None and r["array_field"][0] < 100,
|
||
True, "array not null and [0] < 100"),
|
||
# Multiple fields all not null
|
||
("varchar_field is not null and json_field is not null and array_field is not null",
|
||
lambda r: r["varchar_field"] is not None and r["json_field"] is not None and r["array_field"] is not None,
|
||
True, "multi fields all not null"),
|
||
# Complex: int32 null, float > 0.7, varchar not null
|
||
("int32_field is null and float_field > 0.7 and varchar_field is not null",
|
||
lambda r: (r["int32_field"] is None and
|
||
r["float_field"] is not None and r["float_field"] > 0.7 and
|
||
r["varchar_field"] is not None),
|
||
True, "int32 null and float > 0.7 and varchar not null"),
|
||
# Complex: varchar not null, int64 in [5, 15], float null
|
||
("varchar_field is not null and 5 <= int64_field <= 15 and float_field is null",
|
||
lambda r: (r["varchar_field"] is not None and
|
||
r["int64_field"] is not None and 5 <= r["int64_field"] <= 15 and
|
||
r["float_field"] is None),
|
||
True, "varchar not null and int64 [5,15] and float null"),
|
||
# Complex: int8 not null < 15, double null, varchar not null like varchar_2%
|
||
("int8_field is not null and int8_field < 15 and double_field is null and "
|
||
"varchar_field is not null and varchar_field like \"varchar_2%\"",
|
||
lambda r: (r["int8_field"] is not None and r["int8_field"] < 15 and
|
||
r["double_field"] is None and
|
||
r["varchar_field"] is not None and r["varchar_field"].startswith("varchar_2")),
|
||
True, "int8 < 15 and double null and varchar like varchar_2%"),
|
||
]
|
||
|
||
for filter_str, predicate, with_vec, desc in query_cases:
|
||
expected = [r for r in total_rows if predicate(r)]
|
||
log.info(f"query {desc}: filter={filter_str}, expected={len(expected)}")
|
||
self.query(
|
||
client,
|
||
collection_name,
|
||
filter=filter_str,
|
||
output_fields=['*'],
|
||
check_task=CheckTasks.check_query_results,
|
||
check_items={
|
||
"exp_res": expected,
|
||
"with_vec": with_vec,
|
||
"vector_type": vector_type,
|
||
"pk_name": "id"
|
||
}
|
||
)
|
||
|
||
t1 = time.time()
|
||
log.info(f"Query on all scalar fields cost {t1 - t0:.4f} seconds")
|
||
|
||
# 7. Delete data
|
||
t0 = time.time()
|
||
self.delete(client, collection_name, filter=default_search_exp)
|
||
t1 = time.time()
|
||
log.info(f"Delete cost {t1 - t0:.4f} seconds")
|
||
|
||
# 8. Verify deletion via query
|
||
self.query(
|
||
client,
|
||
collection_name,
|
||
filter=default_search_exp,
|
||
check_task=CheckTasks.check_query_results,
|
||
check_items={"exp_res": []}
|
||
)
|
||
|
||
# 9. Verify deletion via search — should return 0 results
|
||
self.search(
|
||
client,
|
||
collection_name,
|
||
vectors_to_search,
|
||
anns_field="vector",
|
||
search_params=search_params,
|
||
limit=default_limit,
|
||
check_task=CheckTasks.check_search_results,
|
||
check_items={"enable_milvus_client_api": True,
|
||
"nq": len(vectors_to_search),
|
||
"pk_name": "id",
|
||
"limit": 0,
|
||
"metric": "COSINE"}
|
||
)
|
||
|
||
# 10. Cleanup
|
||
self.release_collection(client, collection_name)
|
||
self.drop_collection(client, collection_name)
|
||
|
||
@pytest.mark.tags(CaseLabel.L0)
|
||
@pytest.mark.parametrize("flush_enable", [True, False])
|
||
def test_milvus_client_data_consistent(self, flush_enable):
|
||
"""
|
||
target: verify data consistency between inserted data and query_iterator results
|
||
method: 1. create collection with nullable scalar fields + array fields
|
||
2. insert 6000 rows (2 batches × 3000) with ~20% nulls
|
||
3. create COSINE index, load, search with metric verification
|
||
4. use query_iterator to retrieve all rows
|
||
5. compare query_iterator results with original inserted data (epsilon-aware)
|
||
expected: query_iterator results exactly match inserted data (order-independent, float-epsilon-tolerant)
|
||
"""
|
||
client = self._client()
|
||
dim = 28
|
||
vector_type = DataType.FLOAT_VECTOR
|
||
|
||
# 1. Create collection with custom schema
|
||
collection_name = cf.gen_collection_name_by_testcase_name()
|
||
schema = self.create_schema(client, enable_dynamic_field=False)[0]
|
||
# Primary key and vector field
|
||
schema.add_field("id", DataType.INT64, is_primary=True, auto_id=False)
|
||
schema.add_field("vector", vector_type, dim=dim)
|
||
# Boolean type
|
||
schema.add_field("bool_field", DataType.BOOL, nullable=True)
|
||
# Integer types
|
||
schema.add_field("int16_field", DataType.INT16, nullable=True)
|
||
schema.add_field("int32_field", DataType.INT32, nullable=True)
|
||
schema.add_field("int64_field", DataType.INT64, nullable=True)
|
||
# Float types
|
||
schema.add_field("float_field", DataType.FLOAT, nullable=True)
|
||
schema.add_field("double_field", DataType.DOUBLE, nullable=True)
|
||
# String type
|
||
schema.add_field("varchar_field", DataType.VARCHAR, max_length=200, nullable=True)
|
||
# JSON type
|
||
schema.add_field("json_field", DataType.JSON, nullable=True)
|
||
# Array float type
|
||
schema.add_field("array_float_field", DataType.ARRAY, element_type=DataType.FLOAT, max_capacity=15, nullable=True)
|
||
# Array varchar type
|
||
schema.add_field("array_varchar_field", DataType.ARRAY, element_type=DataType.VARCHAR, max_capacity=15, max_length=100, nullable=True)
|
||
|
||
# Create collection
|
||
self.create_collection(client, collection_name, schema=schema)
|
||
|
||
# 2. Insert data with null values for nullable fields
|
||
num_inserts = 2 # 2 batches to cover sealed + growing scenarios
|
||
total_rows = []
|
||
for i in range(num_inserts):
|
||
data = cf.gen_row_data_by_schema(nb=default_nb, schema=schema, start=i * default_nb)
|
||
self.insert(client, collection_name, data)
|
||
total_rows.extend(data)
|
||
log.info(f"Total inserted {num_inserts * default_nb} entities")
|
||
|
||
if flush_enable:
|
||
self.flush(client, collection_name)
|
||
log.info("Flush enabled: executing flush operation")
|
||
|
||
# Create index parameters
|
||
index_params = self.prepare_index_params(client)[0]
|
||
index_params.add_index("vector", metric_type="COSINE")
|
||
# 3. create index
|
||
self.create_index(client, collection_name, index_params)
|
||
|
||
# 4. Load collection
|
||
self.load_collection(client, collection_name)
|
||
|
||
# 5. Search
|
||
vectors_to_search = cf.gen_vectors(1, dim, vector_data_type=vector_type)
|
||
search_params = {"metric_type": "COSINE", "params": {"nprobe": 100}}
|
||
search_res, _ = self.search(
|
||
client,
|
||
collection_name,
|
||
vectors_to_search,
|
||
anns_field="vector",
|
||
search_params=search_params,
|
||
limit=default_limit,
|
||
output_fields=['*'],
|
||
check_task=CheckTasks.check_search_results,
|
||
check_items={"enable_milvus_client_api": True,
|
||
"nq": len(vectors_to_search),
|
||
"pk_name": "id",
|
||
"limit": default_limit,
|
||
"metric": "COSINE"}
|
||
)
|
||
|
||
# use query iterator to get all the data and compare with the inserted original data
|
||
query_total_rows = []
|
||
query_iterator = self.query_iterator(client, collection_name, output_fields=["*"])[0]
|
||
while True:
|
||
res = query_iterator.next()
|
||
if len(res) == 0:
|
||
log.info("search iteration finished, close")
|
||
query_iterator.close()
|
||
break
|
||
query_total_rows.extend(res)
|
||
|
||
# 6. Query with filters on each scalar field
|
||
t1 = time.time()
|
||
compare_res = pc.compare_lists_with_epsilon_ignore_dict_order(a=query_total_rows, b=total_rows)
|
||
assert compare_res, "query result is not consistent with the inserted original data"
|
||
t2 = time.time()
|
||
log.info(f"Query results compare costs {t2 - t1:.4f} seconds")
|
||
|
||
# 7. Cleanup
|
||
self.release_collection(client, collection_name)
|
||
self.drop_collection(client, collection_name)
|