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
571 lines
22 KiB
Python
571 lines
22 KiB
Python
"""
|
|
CDC sync tests for search and query result verification across vector types.
|
|
"""
|
|
|
|
import random
|
|
import time
|
|
|
|
import pytest
|
|
from pymilvus import AnnSearchRequest, DataType, RRFRanker
|
|
|
|
from .base import TestCDCSyncBase, logger
|
|
|
|
# fmt: off
|
|
VECTOR_PARAMS = [
|
|
("FLOAT_VECTOR", "HNSW", "COSINE", 128),
|
|
("FLOAT_VECTOR", "IVF_FLAT", "L2", 128),
|
|
("FLOAT16_VECTOR", "HNSW", "L2", 64),
|
|
("BFLOAT16_VECTOR", "HNSW", "L2", 64),
|
|
("INT8_VECTOR", "HNSW", "COSINE", 64),
|
|
("BINARY_VECTOR", "BIN_FLAT", "HAMMING", 128),
|
|
("SPARSE_FLOAT_VECTOR", "SPARSE_INVERTED_INDEX", "IP", 0),
|
|
]
|
|
# fmt: on
|
|
|
|
|
|
class TestCDCSyncSearchVerification(TestCDCSyncBase):
|
|
"""Test CDC sync for search and query result verification across vector types."""
|
|
|
|
def setup_method(self):
|
|
"""Setup for each test method."""
|
|
self.resources_to_cleanup = []
|
|
|
|
def teardown_method(self):
|
|
"""Cleanup after each test method - only cleanup upstream, downstream will sync."""
|
|
upstream_client = getattr(self, "_upstream_client", None)
|
|
|
|
if upstream_client:
|
|
for resource_type, resource_name in self.resources_to_cleanup:
|
|
if resource_type == "collection":
|
|
self.cleanup_collection(upstream_client, resource_name)
|
|
|
|
time.sleep(1) # Allow cleanup to sync to downstream
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Internal helper
|
|
# -------------------------------------------------------------------------
|
|
|
|
def _setup_collection(self, client, c_name, vector_type, index_type, metric, dim):
|
|
"""
|
|
Create a single-vector-schema collection, insert 500 records,
|
|
create an index, and load.
|
|
|
|
Returns the collection name (same as c_name).
|
|
"""
|
|
schema = self.create_single_vector_schema(client, vector_type=vector_type, dim=dim)
|
|
client.create_collection(collection_name=c_name, schema=schema)
|
|
|
|
# Insert 500 records
|
|
data = self.generate_single_vector_data(500, vector_type=vector_type, dim=dim)
|
|
client.insert(c_name, data)
|
|
client.flush(c_name)
|
|
|
|
# Build index
|
|
index_params = client.prepare_index_params()
|
|
if index_type == "IVF_FLAT":
|
|
idx_params = {"nlist": 64}
|
|
elif index_type == "HNSW":
|
|
idx_params = {"M": 16, "efConstruction": 200}
|
|
else:
|
|
idx_params = {}
|
|
|
|
index_params.add_index(
|
|
field_name="vector",
|
|
index_type=index_type,
|
|
metric_type=metric,
|
|
params=idx_params,
|
|
)
|
|
client.create_index(c_name, index_params)
|
|
client.load_collection(c_name)
|
|
|
|
return c_name
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Tests
|
|
# -------------------------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize(
|
|
"vector_type,index_type,metric,dim",
|
|
VECTOR_PARAMS,
|
|
ids=[p[0] + "_" + p[1] for p in VECTOR_PARAMS],
|
|
)
|
|
def test_search_result_consistency(
|
|
self,
|
|
upstream_client,
|
|
downstream_client,
|
|
sync_timeout,
|
|
vector_type,
|
|
index_type,
|
|
metric,
|
|
dim,
|
|
):
|
|
"""Verify that ANN search results are consistent between upstream and downstream."""
|
|
start_time = time.time()
|
|
c_name = self.gen_unique_name(f"test_src_{vector_type[:4].lower()}", max_length=50)
|
|
|
|
self.log_test_start(
|
|
"test_search_result_consistency",
|
|
f"SEARCH/{vector_type}/{index_type}",
|
|
c_name,
|
|
)
|
|
self._upstream_client = upstream_client
|
|
self.resources_to_cleanup.append(("collection", c_name))
|
|
|
|
try:
|
|
self.cleanup_collection(upstream_client, c_name)
|
|
self._setup_collection(upstream_client, c_name, vector_type, index_type, metric, dim)
|
|
|
|
# Wait for at least 500 records to appear on downstream
|
|
def check_sync():
|
|
try:
|
|
res = downstream_client.query(
|
|
collection_name=c_name,
|
|
filter="",
|
|
output_fields=["count(*)"],
|
|
)
|
|
cnt = res[0]["count(*)"] if res else 0
|
|
logger.info(f"[SYNC_PROGRESS] downstream count: {cnt}/500")
|
|
return cnt >= 500
|
|
except Exception as e:
|
|
logger.warning(f"Sync check failed: {e}")
|
|
return False
|
|
|
|
assert self.wait_for_sync(check_sync, sync_timeout, f"data sync 500 records {c_name}"), (
|
|
f"Downstream did not receive 500 records within {sync_timeout}s"
|
|
)
|
|
|
|
# Build 5 random query vectors
|
|
dtype = getattr(DataType, vector_type)
|
|
query_vectors = self._gen_vectors(5, dim if dim > 0 else 1000, dtype)
|
|
|
|
avg_overlap, _, _ = self.verify_search_consistency(
|
|
upstream_client,
|
|
downstream_client,
|
|
c_name,
|
|
query_vectors,
|
|
anns_field="vector",
|
|
limit=10,
|
|
metric_type=metric,
|
|
)
|
|
|
|
assert avg_overlap >= self.SEARCH_OVERLAP_THRESHOLD, (
|
|
f"Search overlap {avg_overlap:.4f} is below threshold "
|
|
f"{self.SEARCH_OVERLAP_THRESHOLD} for {vector_type}/{index_type}"
|
|
)
|
|
|
|
finally:
|
|
self.log_test_end(
|
|
"test_search_result_consistency",
|
|
True,
|
|
time.time() - start_time,
|
|
)
|
|
|
|
@pytest.mark.parametrize(
|
|
"vector_type,index_type,metric,dim",
|
|
VECTOR_PARAMS,
|
|
ids=[p[0] + "_" + p[1] for p in VECTOR_PARAMS],
|
|
)
|
|
def test_query_data_sampling(
|
|
self,
|
|
upstream_client,
|
|
downstream_client,
|
|
sync_timeout,
|
|
vector_type,
|
|
index_type,
|
|
metric,
|
|
dim,
|
|
):
|
|
"""Verify scalar field values are identical on both sides via random sampling."""
|
|
start_time = time.time()
|
|
c_name = self.gen_unique_name(f"test_qds_{vector_type[:4].lower()}", max_length=50)
|
|
|
|
self.log_test_start(
|
|
"test_query_data_sampling",
|
|
f"QUERY_SAMPLE/{vector_type}/{index_type}",
|
|
c_name,
|
|
)
|
|
self._upstream_client = upstream_client
|
|
self.resources_to_cleanup.append(("collection", c_name))
|
|
|
|
try:
|
|
self.cleanup_collection(upstream_client, c_name)
|
|
self._setup_collection(upstream_client, c_name, vector_type, index_type, metric, dim)
|
|
|
|
def check_sync():
|
|
try:
|
|
res = downstream_client.query(
|
|
collection_name=c_name,
|
|
filter="",
|
|
output_fields=["count(*)"],
|
|
)
|
|
cnt = res[0]["count(*)"] if res else 0
|
|
logger.info(f"[SYNC_PROGRESS] downstream count: {cnt}/500")
|
|
return cnt >= 500
|
|
except Exception as e:
|
|
logger.warning(f"Sync check failed: {e}")
|
|
return False
|
|
|
|
assert self.wait_for_sync(check_sync, sync_timeout, f"data sync 500 records {c_name}"), (
|
|
f"Downstream did not receive 500 records within {sync_timeout}s"
|
|
)
|
|
|
|
output_fields = ["id", "int_field", "varchar_field", "float_field"]
|
|
match_count, mismatch_count, mismatch_details = self.verify_data_sampling(
|
|
upstream_client,
|
|
downstream_client,
|
|
c_name,
|
|
sample_ratio=0.2,
|
|
output_fields=output_fields,
|
|
)
|
|
|
|
logger.info(
|
|
f"[RESULT] Sampling — match={match_count}, mismatch={mismatch_count}, details={mismatch_details[:3]}"
|
|
)
|
|
assert mismatch_count == 0, f"Found {mismatch_count} mismatched records: {mismatch_details[:5]}"
|
|
|
|
finally:
|
|
self.log_test_end(
|
|
"test_query_data_sampling",
|
|
True,
|
|
time.time() - start_time,
|
|
)
|
|
|
|
def test_hybrid_search_consistency(
|
|
self,
|
|
upstream_client,
|
|
downstream_client,
|
|
sync_timeout,
|
|
):
|
|
"""Verify hybrid search (dense + sparse, RRF ranker) results are consistent."""
|
|
start_time = time.time()
|
|
c_name = self.gen_unique_name("test_hybrid_srch", max_length=50)
|
|
|
|
self.log_test_start("test_hybrid_search_consistency", "HYBRID_SEARCH", c_name)
|
|
self._upstream_client = upstream_client
|
|
self.resources_to_cleanup.append(("collection", c_name))
|
|
|
|
try:
|
|
self.cleanup_collection(upstream_client, c_name)
|
|
|
|
# Build schema: dense FloatVector(128) + sparse
|
|
schema = upstream_client.create_schema(enable_dynamic_field=True)
|
|
schema.add_field("id", DataType.INT64, is_primary=True, auto_id=True)
|
|
schema.add_field("dense", DataType.FLOAT_VECTOR, dim=128)
|
|
schema.add_field("sparse", DataType.SPARSE_FLOAT_VECTOR)
|
|
schema.add_field("int_field", DataType.INT64)
|
|
schema.add_field("varchar_field", DataType.VARCHAR, max_length=256)
|
|
|
|
upstream_client.create_collection(collection_name=c_name, schema=schema)
|
|
|
|
# Insert 300 records
|
|
dense_vecs = self._gen_vectors(300, 128, DataType.FLOAT_VECTOR)
|
|
sparse_vecs = self._gen_vectors(300, 1000, DataType.SPARSE_FLOAT_VECTOR)
|
|
data = [
|
|
{
|
|
"dense": dense_vecs[i],
|
|
"sparse": sparse_vecs[i],
|
|
"int_field": random.randint(0, 1000),
|
|
"varchar_field": f"hybrid_{i}_{random.randint(1000, 9999)}",
|
|
}
|
|
for i in range(300)
|
|
]
|
|
upstream_client.insert(c_name, data)
|
|
upstream_client.flush(c_name)
|
|
|
|
# Create indexes
|
|
index_params = upstream_client.prepare_index_params()
|
|
index_params.add_index(
|
|
field_name="dense",
|
|
index_type="HNSW",
|
|
metric_type="COSINE",
|
|
params={"M": 16, "efConstruction": 200},
|
|
)
|
|
index_params.add_index(
|
|
field_name="sparse",
|
|
index_type="SPARSE_INVERTED_INDEX",
|
|
metric_type="IP",
|
|
params={},
|
|
)
|
|
upstream_client.create_index(c_name, index_params)
|
|
upstream_client.load_collection(c_name)
|
|
|
|
# Wait for downstream sync
|
|
def check_sync():
|
|
try:
|
|
res = downstream_client.query(
|
|
collection_name=c_name,
|
|
filter="",
|
|
output_fields=["count(*)"],
|
|
)
|
|
cnt = res[0]["count(*)"] if res else 0
|
|
logger.info(f"[SYNC_PROGRESS] downstream count: {cnt}/300")
|
|
return cnt >= 300
|
|
except Exception as e:
|
|
logger.warning(f"Sync check failed: {e}")
|
|
return False
|
|
|
|
assert self.wait_for_sync(check_sync, sync_timeout, f"hybrid data sync {c_name}"), (
|
|
f"Downstream did not receive 300 records within {sync_timeout}s"
|
|
)
|
|
|
|
# Build hybrid search requests
|
|
q_dense = self._gen_vectors(1, 128, DataType.FLOAT_VECTOR)[0]
|
|
q_sparse = self._gen_vectors(1, 1000, DataType.SPARSE_FLOAT_VECTOR)[0]
|
|
|
|
dense_req = AnnSearchRequest(
|
|
data=[q_dense],
|
|
anns_field="dense",
|
|
param={"metric_type": "COSINE", "params": {"ef": 64}},
|
|
limit=10,
|
|
)
|
|
sparse_req = AnnSearchRequest(
|
|
data=[q_sparse],
|
|
anns_field="sparse",
|
|
param={"metric_type": "IP"},
|
|
limit=10,
|
|
)
|
|
|
|
up_results = upstream_client.hybrid_search(
|
|
collection_name=c_name,
|
|
reqs=[dense_req, sparse_req],
|
|
ranker=RRFRanker(),
|
|
limit=10,
|
|
output_fields=["id"],
|
|
)
|
|
down_results = downstream_client.hybrid_search(
|
|
collection_name=c_name,
|
|
reqs=[dense_req, sparse_req],
|
|
ranker=RRFRanker(),
|
|
limit=10,
|
|
output_fields=["id"],
|
|
)
|
|
|
|
up_pks = set(hit["id"] for hit in up_results[0]) if up_results else set()
|
|
down_pks = set(hit["id"] for hit in down_results[0]) if down_results else set()
|
|
union_size = len(up_pks | down_pks)
|
|
overlap = len(up_pks & down_pks) / union_size if union_size > 0 else 1.0
|
|
|
|
logger.info(f"[RESULT] Hybrid search PK overlap={overlap:.4f} (up={len(up_pks)}, down={len(down_pks)})")
|
|
assert overlap >= self.SEARCH_OVERLAP_THRESHOLD, (
|
|
f"Hybrid search overlap {overlap:.4f} below threshold {self.SEARCH_OVERLAP_THRESHOLD}"
|
|
)
|
|
|
|
finally:
|
|
self.log_test_end(
|
|
"test_hybrid_search_consistency",
|
|
True,
|
|
time.time() - start_time,
|
|
)
|
|
|
|
def test_search_iterator_consistency(
|
|
self,
|
|
upstream_client,
|
|
downstream_client,
|
|
sync_timeout,
|
|
):
|
|
"""Verify search iterator returns the same PK set on upstream and downstream."""
|
|
start_time = time.time()
|
|
c_name = self.gen_unique_name("test_srch_iter", max_length=50)
|
|
|
|
self.log_test_start("test_search_iterator_consistency", "SEARCH_ITERATOR", c_name)
|
|
self._upstream_client = upstream_client
|
|
self.resources_to_cleanup.append(("collection", c_name))
|
|
|
|
try:
|
|
self.cleanup_collection(upstream_client, c_name)
|
|
self._setup_collection(upstream_client, c_name, "FLOAT_VECTOR", "HNSW", "COSINE", 128)
|
|
|
|
def check_sync():
|
|
try:
|
|
res = downstream_client.query(
|
|
collection_name=c_name,
|
|
filter="",
|
|
output_fields=["count(*)"],
|
|
)
|
|
cnt = res[0]["count(*)"] if res else 0
|
|
logger.info(f"[SYNC_PROGRESS] downstream count: {cnt}/500")
|
|
return cnt >= 500
|
|
except Exception as e:
|
|
logger.warning(f"Sync check failed: {e}")
|
|
return False
|
|
|
|
assert self.wait_for_sync(check_sync, sync_timeout, f"data sync 500 records {c_name}"), (
|
|
f"Downstream did not receive 500 records within {sync_timeout}s"
|
|
)
|
|
|
|
query_vec = self._gen_vectors(1, 128, DataType.FLOAT_VECTOR)[0]
|
|
search_params = {"metric_type": "COSINE", "params": {"ef": 64}}
|
|
|
|
def _collect_iterator_pks(client):
|
|
pks = set()
|
|
iterator = client.search_iterator(
|
|
collection_name=c_name,
|
|
data=[query_vec],
|
|
anns_field="vector",
|
|
batch_size=50,
|
|
limit=200,
|
|
param=search_params,
|
|
output_fields=["id"],
|
|
)
|
|
while True:
|
|
batch = iterator.next()
|
|
if not batch:
|
|
iterator.close()
|
|
break
|
|
for hit in batch:
|
|
pks.add(hit["id"])
|
|
return pks
|
|
|
|
up_pks = _collect_iterator_pks(upstream_client)
|
|
down_pks = _collect_iterator_pks(downstream_client)
|
|
union_size = len(up_pks | down_pks)
|
|
overlap = len(up_pks & down_pks) / union_size if union_size > 0 else 1.0
|
|
|
|
logger.info(f"[RESULT] Search iterator overlap={overlap:.4f} (up={len(up_pks)}, down={len(down_pks)})")
|
|
assert overlap >= self.SEARCH_OVERLAP_THRESHOLD, (
|
|
f"Search iterator PK overlap {overlap:.4f} below threshold {self.SEARCH_OVERLAP_THRESHOLD}"
|
|
)
|
|
|
|
finally:
|
|
self.log_test_end(
|
|
"test_search_iterator_consistency",
|
|
True,
|
|
time.time() - start_time,
|
|
)
|
|
|
|
def test_query_iterator_consistency(
|
|
self,
|
|
upstream_client,
|
|
downstream_client,
|
|
sync_timeout,
|
|
):
|
|
"""Verify that a query iterator retrieves identical PK sets from both sides."""
|
|
start_time = time.time()
|
|
c_name = self.gen_unique_name("test_qry_iter", max_length=50)
|
|
|
|
self.log_test_start("test_query_iterator_consistency", "QUERY_ITERATOR", c_name)
|
|
self._upstream_client = upstream_client
|
|
self.resources_to_cleanup.append(("collection", c_name))
|
|
|
|
try:
|
|
self.cleanup_collection(upstream_client, c_name)
|
|
self._setup_collection(upstream_client, c_name, "FLOAT_VECTOR", "HNSW", "COSINE", 128)
|
|
|
|
def check_sync():
|
|
try:
|
|
res = downstream_client.query(
|
|
collection_name=c_name,
|
|
filter="",
|
|
output_fields=["count(*)"],
|
|
)
|
|
cnt = res[0]["count(*)"] if res else 0
|
|
logger.info(f"[SYNC_PROGRESS] downstream count: {cnt}/500")
|
|
return cnt >= 500
|
|
except Exception as e:
|
|
logger.warning(f"Sync check failed: {e}")
|
|
return False
|
|
|
|
assert self.wait_for_sync(check_sync, sync_timeout, f"data sync 500 records {c_name}"), (
|
|
f"Downstream did not receive 500 records within {sync_timeout}s"
|
|
)
|
|
|
|
up_count, down_count, match = self.verify_iterator_consistency(
|
|
upstream_client,
|
|
downstream_client,
|
|
c_name,
|
|
batch_size=100,
|
|
)
|
|
|
|
logger.info(f"[RESULT] Query iterator — upstream={up_count}, downstream={down_count}, match={match}")
|
|
assert match, f"Query iterator PK sets differ: upstream={up_count}, downstream={down_count}"
|
|
|
|
finally:
|
|
self.log_test_end(
|
|
"test_query_iterator_consistency",
|
|
True,
|
|
time.time() - start_time,
|
|
)
|
|
|
|
def test_search_with_filter_consistency(
|
|
self,
|
|
upstream_client,
|
|
downstream_client,
|
|
sync_timeout,
|
|
):
|
|
"""Verify filtered search produces consistent results honoring the filter predicate."""
|
|
start_time = time.time()
|
|
c_name = self.gen_unique_name("test_srch_filter", max_length=50)
|
|
|
|
self.log_test_start("test_search_with_filter_consistency", "SEARCH_WITH_FILTER", c_name)
|
|
self._upstream_client = upstream_client
|
|
self.resources_to_cleanup.append(("collection", c_name))
|
|
|
|
try:
|
|
self.cleanup_collection(upstream_client, c_name)
|
|
self._setup_collection(upstream_client, c_name, "FLOAT_VECTOR", "HNSW", "COSINE", 128)
|
|
|
|
def check_sync():
|
|
try:
|
|
res = downstream_client.query(
|
|
collection_name=c_name,
|
|
filter="",
|
|
output_fields=["count(*)"],
|
|
)
|
|
cnt = res[0]["count(*)"] if res else 0
|
|
logger.info(f"[SYNC_PROGRESS] downstream count: {cnt}/500")
|
|
return cnt >= 500
|
|
except Exception as e:
|
|
logger.warning(f"Sync check failed: {e}")
|
|
return False
|
|
|
|
assert self.wait_for_sync(check_sync, sync_timeout, f"data sync 500 records {c_name}"), (
|
|
f"Downstream did not receive 500 records within {sync_timeout}s"
|
|
)
|
|
|
|
filter_expr = "int_field > 500"
|
|
query_vec = self._gen_vectors(1, 128, DataType.FLOAT_VECTOR)[0]
|
|
search_params = {"metric_type": "COSINE"}
|
|
|
|
up_results = upstream_client.search(
|
|
collection_name=c_name,
|
|
data=[query_vec],
|
|
anns_field="vector",
|
|
search_params=search_params,
|
|
filter=filter_expr,
|
|
limit=10,
|
|
output_fields=["id", "int_field"],
|
|
)
|
|
down_results = downstream_client.search(
|
|
collection_name=c_name,
|
|
data=[query_vec],
|
|
anns_field="vector",
|
|
search_params=search_params,
|
|
filter=filter_expr,
|
|
limit=10,
|
|
output_fields=["id", "int_field"],
|
|
)
|
|
|
|
# Verify filter is honoured on both sides
|
|
for hit in up_results[0] if up_results else []:
|
|
assert hit["int_field"] > 500, f"Filter violated on upstream: int_field={hit['int_field']}"
|
|
for hit in down_results[0] if down_results else []:
|
|
assert hit["int_field"] > 500, f"Filter violated on downstream: int_field={hit['int_field']}"
|
|
|
|
# Verify PK overlap
|
|
up_pks = set(hit["id"] for hit in up_results[0]) if up_results else set()
|
|
down_pks = set(hit["id"] for hit in down_results[0]) if down_results else set()
|
|
union_size = len(up_pks | down_pks)
|
|
overlap = len(up_pks & down_pks) / union_size if union_size > 0 else 1.0
|
|
|
|
logger.info(f"[RESULT] Filtered search overlap={overlap:.4f} (up={len(up_pks)}, down={len(down_pks)})")
|
|
assert overlap >= self.SEARCH_OVERLAP_THRESHOLD, (
|
|
f"Filtered search overlap {overlap:.4f} below threshold {self.SEARCH_OVERLAP_THRESHOLD}"
|
|
)
|
|
|
|
finally:
|
|
self.log_test_end(
|
|
"test_search_with_filter_consistency",
|
|
True,
|
|
time.time() - start_time,
|
|
)
|