mirror of
https://github.com/milvus-io/milvus.git
synced 2026-07-21 18:25:44 +00:00
e2787d3981
issue: #47420 ## What this PR does Project-wide migration of raw `fmt.Errorf` / `errors.New` in function bodies onto the `merr` framework, plus the Sys-vs-Input error classification and the machinery it drives (retriability, fine-grained metrics, segcore unification), plus the convention docs and a linter that keeps it from regressing. Scope: storage, proxy, coordinators (root/data/query), query node, data node, `pkg/util` & `internal/util`, expression parser, message queue, streaming, and misc packages. Bare raw-error usages went from ~3000 to a ~340 allowlist (package-level sentinels / build-tag / test sites). --- ## How to review this PR It is large but the vast majority is mechanical. Changes fall into three tiers; spend review budget on Part 2 and Part 3. ### Part 1 — Mechanical standardization (low risk, verify by rule) Each converted call follows one of a small fixed set of rules. To review, check that each site obeys the matching rule rather than reading every line: | Pattern | Rule | |---|---| | `fmt.Errorf("...")` originating a new error | → `merr.WrapErrXxxMsg("...")` with a code matching the failure's meaning | | Adding context to an existing typed error | → `merr.Wrap(err, "...")` / `merr.Wrapf(...)` — **preserves** the inner code (never `WrapErr*Err`, which overwrites it) | | Errors inside the streaming subsystem | → `status.New*` factories (StreamingError), **not** merr — this is the component-internal dialect (see `docs/dev/error_handling_guide.md`) | | Low-level / control-flow signal caught by `errors.Is` | → kept as a package-level `errors.New` sentinel (lowercase, same-package) | Conventions are documented in `docs/dev/error_handling_guide.md` (how-to) and `docs/dev/error_sentinel_convention.md` (rules + audit). A `gocritic`/`ruleguard` rule (`rawmerrerror`, in `rules.go`) enforces "no raw `return errors.New/fmt.Errorf`" under `make verifiers`. ### Part 2 — Behavior changes (review these closely) These are the sites where the wire contract or runtime behavior changes, not just the source text. Listed by category; representative locations given, full set in the diff. **A. gRPC wire-code shifts: `UnexpectedError(1)/Code 65535` → typed code.** Where a handler previously returned a raw error (collapsed to `Code=65535` on the wire), it now returns a typed merr, so the client sees a real code. The most common shift is to `IllegalArgument(5)/Code 1100` (ParameterInvalid). Touch points include datanode task handlers (CreateTask/Query/Drop), proxy Upsert, querynode GetMetrics, datacoord CreateIndex, httpserver query-response builder, and typeutil schema validation. One code refinement: an index-param validation moved `1100` → `1101` (ParameterMissing). **Client/SDK assertions and any code that switched on `Code=65535` for these paths must be re-checked** (the go_client e2e assertions were already aligned in this PR). **B. Prometheus `status` label contract change (externally visible).** The proxy metric's coarse `fail` / `rejected` values are split into `fail_input` / `fail_system` and `rejected_user` / `rejected_system` (in `requestutil.ParseMetricLabel`; auth/privilege rejections count as `rejected_user`), so dashboards can attribute a failure to caller vs operator. **Dashboards/alerts querying `status="fail"` must migrate to `status=~"fail_.*"`, and `status="rejected"` to `status=~"rejected_.*"`.** The in-repo Grafana dashboard is already migrated; external dashboards built on the old values silently go empty after upgrade. This is the one change that requires an ops-side migration. **C. Retriability semantics.** - C1: `merr.Status(err)` now forces `Retriable=false` when the error is an `InputError` — a malformed request can never succeed on blind retry, so clients never get the self-contradictory "your input is wrong but you may retry". - C2: `retry.Do` short-circuits an `InputError` (non-retriable) — **but only when the caller did not pass a `RetryErr` predicate**. The check is an `if c.isRetryErr != nil { ... } else if InputError { ... }` *mutually exclusive* branch (`pkg/util/retry/retry.go`): an explicit `RetryErr` takes precedence and bypasses the InputError abort. `retry.Handle` deliberately does **not** apply the InputError abort (its callers signal abort via `shouldRetry=false`). Four flusher startup callsites that must retry through transient "not ready" errors were given explicit `RetryErr` escape hatches. **D. segcore (C++→Go) error classification.** A single shared Go-side table (`pkg/util/merr/segcore.go`) maps each segcore code to a merr sentinel + InputError/signal category, replacing scattered hand-written `if errorCode == ...` switches in the cgo wrappers. **Wire `Code` values change for every segcore pass-through error, not just the remapped ones.** Named sentinels remap (C++ `2003` → merr `2001`, `2033` → `2002`, Folly/Knowhere codes likewise); **all remaining pass-through codes (`2004`–`2043`, previously surfaced to clients as raw C++ enum values) now serialize as `2000`** (`ErrSegcore`), with the original C++ code preserved in the `Reason` text (`segcoreCode=...`); unknown/future codes collapse to `2000` as well (pinned by the `wire_code_projection` test). Transient segcore classes (object storage / file IO / OOM / mmap / FieldNotLoaded — 11 codes) now report `Retriable=true`. **Any client switching on raw segcore codes in the `2004`–`2043` range must be re-checked**; the in-Reason code remains available for diagnostics. Signal codes (PretendFinished / FollyCancel) are recognized centrally. `errors.Is`-based control flow on these (e.g. scheduler skip/retry) is preserved. **E. InputError classification (25 sentinels + dynamic marks).** 25 sentinels in `errors.go` carry `WithErrorType(InputError)` (the Collection / ResourceGroup / Database families, `ErrIndexDuplicate`, `ErrParameterInvalid`, `ErrPrivilegeNotAuthenticated`, `ErrImportFailed`, `ErrQueryPlan`, ...), plus dynamic marks for the 8 segcore input codes (ExprInvalid, DimNotMatch, MetricTypeInvalid, FieldIDInvalid, ...) and `WrapErrAsInputError`. The widest blast radius is `ErrParameterInvalid` (1100): ~2335 `WrapErrParameterInvalid*` callsites now classify as input / non-retriable. Because of C1/C2 this changes retriability for any path that returns these. **The audit to confirm no transient path was mis-marked is the single most important review item** (see Part 3). One reverse correction: storage field-stats parsing moved from `ErrParameterInvalid` (input) to `ErrDataIntegrity` — a corrupted stored stat is data corruption, not user input. ### Part 3 — Known risks & traps (called out proactively) 1. **`merr.Wrap` vs `WrapErr*Err` (code-masking).** `WrapErr*Err` builds a `wrappedMilvusError{sentinel: ErrServiceInternal}` whose `code()` returns the *outer* sentinel — it overwrites the inner typed code and hides the `errors.Is` chain. This is intentional (use it to *deliberately* downgrade), but it was a recurring conversion defect; the rule "add context with `merr.Wrap`, downgrade with `WrapErr*Err`" is enforced by convention and reviewed across the diff. 2. **InputError × `retry.Do` blast radius.** Marking a sentinel `InputError` makes any `retry.Do(...)` without a `RetryErr` predicate stop retrying it. Reviewers should sanity-check that no transient use of the 19 newly-marked sentinels (especially `ErrParameterInvalid`) sits inside a retry loop that needed to keep spinning. The known flusher cases were handled (see C2). 3. **The ~340 raw-error allowlist.** What remains as bare `errors.New` is, by design: package-level sentinels (caught by `errors.Is`), `//go:build test` sites, and out-of-band trees (`cmd/`, `tests/`, codegen, walimpls). The linter only bans the *direct-return* form; assignment-then-return escapes and the full no-exceptions ban are deferred to an AST-based linter (Tier 2, documented). 4. **segcore C++ second step deferred.** This PR unifies classification on the Go side; splitting the dual-semantic C++ codes at the source is a follow-up. --- ## Validation - `make verifiers`: Go side clean (gofmt + static-check across modules, including the new `rawmerrerror` rule with a 0-hit baseline repo-wide). - `make test-go`: passing; the one real regression introduced (a datanode `invalid_task_type` assertion shifting `1` → `5` from a ParameterInvalid conversion) was fixed in-tree. - go_client e2e CreateIndex assertions aligned to the new merr messages. --------- Signed-off-by: zhenshan.cao <zhenshan.cao@zilliz.com> Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2081 lines
93 KiB
Python
2081 lines
93 KiB
Python
import threading
|
|
import time
|
|
|
|
import pytest
|
|
from pymilvus import DefaultConfig
|
|
from pymilvus.exceptions import MilvusException
|
|
from base.client_base import TestcaseBase
|
|
from base.collection_wrapper import ApiCollectionWrapper
|
|
from base.utility_wrapper import ApiUtilityWrapper
|
|
from common.common_params import FieldParams, DefaultVectorIndexParams, DefaultVectorSearchParams
|
|
from utils.util_log import test_log as log
|
|
from common import common_func as cf
|
|
from common import common_type as ct
|
|
from common.common_type import CaseLabel, CheckTasks
|
|
from common.milvus_sys import MilvusSys
|
|
from pymilvus.grpc_gen.common_pb2 import SegmentState
|
|
import random
|
|
from pymilvus.client.types import ResourceGroupConfig
|
|
import copy
|
|
|
|
prefix = "utility"
|
|
default_schema = cf.gen_default_collection_schema()
|
|
default_int64_field_name = ct.default_int64_field_name
|
|
default_field_name = ct.default_float_vec_field_name
|
|
default_index_params = {"index_type": "IVF_SQ8", "metric_type": "L2", "params": {"nlist": 64}}
|
|
default_dim = ct.default_dim
|
|
default_nb = ct.default_nb
|
|
num_loaded_entities = "num_loaded_entities"
|
|
num_total_entities = "num_total_entities"
|
|
loading_progress = "loading_progress"
|
|
num_loaded_partitions = "num_loaded_partitions"
|
|
not_loaded_partitions = "not_loaded_partitions"
|
|
exp_name = "name"
|
|
exp_schema = "schema"
|
|
|
|
|
|
class TestUtilityParams(TestcaseBase):
|
|
""" Test case of index interface """
|
|
|
|
@pytest.fixture(scope="function", params=["JACCARD", "Superstructure", "Substructure"])
|
|
def get_not_support_metric(self, request):
|
|
yield request.param
|
|
|
|
@pytest.fixture(scope="function", params=["metric_type", "metric"])
|
|
def get_support_metric_field(self, request):
|
|
yield request.param
|
|
|
|
@pytest.fixture(scope="function", params=ct.get_not_string)
|
|
def get_invalid_type_collection_name(self, request):
|
|
yield request.param
|
|
|
|
@pytest.fixture(scope="function", params=ct.invalid_resource_names)
|
|
def get_invalid_value_collection_name(self, request):
|
|
yield request.param
|
|
|
|
"""
|
|
******************************************************************
|
|
# The followings are invalid cases
|
|
******************************************************************
|
|
"""
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_has_collection_name_type_invalid(self, get_invalid_type_collection_name):
|
|
"""
|
|
target: test has_collection with error collection name
|
|
method: input invalid name
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
c_name = get_invalid_type_collection_name
|
|
self.utility_wrap.has_collection(c_name, check_task=CheckTasks.err_res,
|
|
check_items={ct.err_code: 999,
|
|
ct.err_msg: f"`collection_name` value {c_name} is illegal"})
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_has_collection_name_value_invalid(self, get_invalid_value_collection_name):
|
|
"""
|
|
target: test has_collection with error collection name
|
|
method: input invalid name
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
c_name = get_invalid_value_collection_name
|
|
error = {ct.err_code: 999, ct.err_msg: f"Invalid collection name: {c_name}"}
|
|
if c_name in [None, ""]:
|
|
error = {ct.err_code: 999, ct.err_msg: f"`collection_name` value {c_name} is illegal"}
|
|
self.utility_wrap.has_collection(c_name, check_task=CheckTasks.err_res, check_items=error)
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_has_partition_collection_name_type_invalid(self, get_invalid_type_collection_name):
|
|
"""
|
|
target: test has_partition with error collection name
|
|
method: input invalid name
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
c_name = get_invalid_type_collection_name
|
|
p_name = cf.gen_unique_str(prefix)
|
|
self.utility_wrap.has_partition(c_name, p_name, check_task=CheckTasks.err_res,
|
|
check_items={ct.err_code: 999,
|
|
ct.err_msg: f"`collection_name` value {c_name} is illegal"})
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_has_partition_collection_name_value_invalid(self, get_invalid_value_collection_name):
|
|
"""
|
|
target: test has_partition with error collection name
|
|
method: input invalid name
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
c_name = get_invalid_value_collection_name
|
|
p_name = cf.gen_unique_str(prefix)
|
|
error = {ct.err_code: 999, ct.err_msg: f"Invalid collection name: {c_name}"}
|
|
if c_name in [None, ""]:
|
|
error = {ct.err_code: 999, ct.err_msg: f"`collection_name` value {c_name} is illegal"}
|
|
self.utility_wrap.has_partition(c_name, p_name, check_task=CheckTasks.err_res, check_items=error)
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_has_partition_name_type_invalid(self, get_invalid_type_collection_name):
|
|
"""
|
|
target: test has_partition with error partition name
|
|
method: input invalid name
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
ut = ApiUtilityWrapper()
|
|
c_name = cf.gen_unique_str(prefix)
|
|
p_name = get_invalid_type_collection_name
|
|
ut.has_partition(c_name, p_name, check_task=CheckTasks.err_res,
|
|
check_items={ct.err_code: 999,
|
|
ct.err_msg: f"`partition_name` value {p_name} is illegal"})
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_has_partition_name_value_invalid(self, get_invalid_value_collection_name):
|
|
"""
|
|
target: test has_partition with error partition name
|
|
method: input invalid name
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
ut = ApiUtilityWrapper()
|
|
c_name = cf.gen_unique_str(prefix)
|
|
p_name = get_invalid_value_collection_name
|
|
if p_name == "12name":
|
|
pytest.skip("partition name 12name is legal")
|
|
if p_name == "n-ame":
|
|
pytest.skip("https://github.com/milvus-io/milvus/issues/39432")
|
|
error = {ct.err_code: 999, ct.err_msg: f"Invalid partition name: {p_name}"}
|
|
if p_name in [None]:
|
|
error = {ct.err_code: 999, ct.err_msg: f"`partition_name` value {p_name} is illegal"}
|
|
elif p_name in [" ", ""]:
|
|
error = {ct.err_code: 999, ct.err_msg: "Invalid partition name: . Partition name should not be empty."}
|
|
ut.has_partition(c_name, p_name, check_task=CheckTasks.err_res, check_items=error)
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_drop_collection_name_type_invalid(self, get_invalid_type_collection_name):
|
|
self._connect()
|
|
c_name = get_invalid_type_collection_name
|
|
self.utility_wrap.drop_collection(c_name, check_task=CheckTasks.err_res,
|
|
check_items={ct.err_code: 999,
|
|
ct.err_msg: f"`collection_name` value {c_name} is illegal"})
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_drop_collection_name_value_invalid(self, get_invalid_value_collection_name):
|
|
self._connect()
|
|
c_name = get_invalid_value_collection_name
|
|
if c_name in [None, ""]:
|
|
error = {ct.err_code: 999, ct.err_msg: f"`collection_name` value {c_name} is illegal"}
|
|
self.utility_wrap.drop_collection(c_name, check_task=CheckTasks.err_res, check_items=error)
|
|
else:
|
|
self.utility_wrap.drop_collection(c_name)
|
|
|
|
# TODO: enable
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_list_collections_using_invalid(self):
|
|
"""
|
|
target: test list_collections with invalid using
|
|
method: input invalid name
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
using = "empty"
|
|
ut = ApiUtilityWrapper()
|
|
ex, _ = ut.list_collections(using=using, check_task=CheckTasks.err_res,
|
|
check_items={ct.err_code: 0, ct.err_msg: "should create connect"})
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
@pytest.mark.parametrize("invalid_name", ct.invalid_resource_names)
|
|
def test_index_process_invalid_name(self, invalid_name):
|
|
"""
|
|
target: test building_process
|
|
method: input invalid name
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
error = {ct.err_code: 999, ct.err_msg: f"Invalid collection name: {invalid_name}"}
|
|
if invalid_name in [None, ""]:
|
|
error = {ct.err_code: 999, ct.err_msg: "collection name should not be empty"}
|
|
self.utility_wrap.index_building_progress(collection_name=invalid_name,
|
|
check_task=CheckTasks.err_res, check_items=error)
|
|
|
|
# TODO: not support index name
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
@pytest.mark.parametrize("invalid_index_name", ct.invalid_resource_names)
|
|
def test_index_process_invalid_index_name(self, invalid_index_name):
|
|
"""
|
|
target: test building_process
|
|
method: input invalid index name
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
collection_w = self.init_collection_wrap()
|
|
error = {ct.err_code: 999, ct.err_msg: "index not found"}
|
|
self.utility_wrap.index_building_progress(collection_name=collection_w.name, index_name=invalid_index_name,
|
|
check_task=CheckTasks.err_res, check_items=error)
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
@pytest.mark.skip("not ready")
|
|
def test_wait_index_invalid_name(self, get_invalid_type_collection_name):
|
|
"""
|
|
target: test wait_index
|
|
method: input invalid name
|
|
expected: raise exception
|
|
"""
|
|
pass
|
|
# self._connect()
|
|
# c_name = get_invalid_collection_name
|
|
# ut = ApiUtilityWrapper()
|
|
# if isinstance(c_name, str) and c_name:
|
|
# ex, _ = ut.wait_for_index_building_complete(c_name,
|
|
# check_items={ct.err_code: 1,
|
|
# ct.err_msg: "Invalid collection name"})
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def _test_wait_index_invalid_index_name(self, get_invalid_index_name):
|
|
"""
|
|
target: test wait_index
|
|
method: input invalid index name
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
c_name = cf.gen_unique_str(prefix)
|
|
index_name = get_invalid_index_name
|
|
ut = ApiUtilityWrapper()
|
|
ex, _ = ut.wait_for_index_building_complete(c_name, index_name)
|
|
log.error(str(ex))
|
|
assert "invalid" or "illegal" in str(ex)
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
@pytest.mark.parametrize("invalid_c_name", ["12-s", "12 s", "(mn)", "中文", "%$#"])
|
|
def test_loading_progress_invalid_collection_name(self, invalid_c_name):
|
|
"""
|
|
target: test loading progress with invalid collection name
|
|
method: input invalid collection name
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
c_name = cf.gen_unique_str(prefix)
|
|
df = cf.gen_default_dataframe_data()
|
|
self.collection_wrap.construct_from_dataframe(c_name, df, primary_field=ct.default_int64_field_name)
|
|
self.collection_wrap.create_index(ct.default_float_vec_field_name, index_params=ct.default_flat_index)
|
|
self.collection_wrap.load()
|
|
error = {ct.err_code: 1, ct.err_msg: "Invalid collection name: {}".format(invalid_c_name)}
|
|
self.utility_wrap.loading_progress(invalid_c_name, check_task=CheckTasks.err_res, check_items=error)
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_loading_progress_not_existed_collection_name(self):
|
|
"""
|
|
target: test loading progress with invalid collection name
|
|
method: input invalid collection name
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
c_name = cf.gen_unique_str(prefix)
|
|
df = cf.gen_default_dataframe_data()
|
|
self.collection_wrap.construct_from_dataframe(c_name, df, primary_field=ct.default_int64_field_name)
|
|
self.collection_wrap.create_index(ct.default_float_vec_field_name, index_params=ct.default_flat_index)
|
|
self.collection_wrap.load()
|
|
error = {ct.err_code: 100, ct.err_msg: "collection not found[database=default][collection=not_existed_name]"}
|
|
self.utility_wrap.loading_progress("not_existed_name", check_task=CheckTasks.err_res, check_items=error)
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
@pytest.mark.parametrize("partition_name", ct.invalid_resource_names)
|
|
def test_loading_progress_invalid_partition_names(self, partition_name):
|
|
"""
|
|
target: test loading progress with invalid partition names
|
|
method: input invalid partition names
|
|
expected: raise an exception
|
|
"""
|
|
if partition_name == "":
|
|
pytest.skip("https://github.com/milvus-io/milvus/issues/38223")
|
|
collection_w = self.init_collection_general(prefix, nb=10)[0]
|
|
partition_names = [partition_name]
|
|
collection_w.load()
|
|
err_msg = {ct.err_code: 999, ct.err_msg: "partition not found"}
|
|
if partition_name is None:
|
|
err_msg = {ct.err_code: 999, ct.err_msg: "is illegal"}
|
|
self.utility_wrap.loading_progress(collection_w.name, partition_names,
|
|
check_task=CheckTasks.err_res, check_items=err_msg)
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
@pytest.mark.parametrize("partition_names", [[ct.default_tag], [ct.default_partition_name, ct.default_tag]])
|
|
def test_loading_progress_not_existed_partitions(self, partition_names):
|
|
"""
|
|
target: test loading progress with not existed partitions
|
|
method: input all or part not existed partition names
|
|
expected: raise exception
|
|
"""
|
|
collection_w = self.init_collection_general(prefix, nb=10)[0]
|
|
collection_w.load()
|
|
err_msg = {ct.err_code: 15, ct.err_msg: f"partition not found"}
|
|
self.utility_wrap.loading_progress(collection_w.name, partition_names,
|
|
check_task=CheckTasks.err_res, check_items=err_msg)
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_wait_for_loading_collection_not_existed(self):
|
|
"""
|
|
target: test wait for loading
|
|
method: input collection not created before
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
c_name = cf.gen_unique_str(prefix)
|
|
self.utility_wrap.wait_for_loading_complete(
|
|
c_name,
|
|
check_task=CheckTasks.err_res,
|
|
check_items={ct.err_code: 100, ct.err_msg: "collection not found[database=default]"
|
|
"[collection={}]".format(c_name)})
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_wait_for_loading_partition_not_existed(self):
|
|
"""
|
|
target: test wait for loading
|
|
method: input partition not created before
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
collection_w = self.init_collection_wrap()
|
|
self.utility_wrap.wait_for_loading_complete(
|
|
collection_w.name, partition_names=[ct.default_tag],
|
|
check_task=CheckTasks.err_res,
|
|
check_items={ct.err_code: 200, ct.err_msg: f'partition not found[partition={ct.default_tag}]'})
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_drop_collection_not_existed(self):
|
|
"""
|
|
target: test drop an not existed collection
|
|
method: drop a not created collection
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
c_name = cf.gen_unique_str(prefix)
|
|
|
|
# error = {ct.err_code: 1, ct.err_msg: f"DescribeCollection failed: can't find collection: {c_name}"}
|
|
# self.utility_wrap.drop_collection(c_name, check_task=CheckTasks.err_res, check_items=error)
|
|
|
|
# @longjiquan: dropping collection should be idempotent.
|
|
self.utility_wrap.drop_collection(c_name)
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
@pytest.mark.skip(reason="calc_distance interface is no longer supported")
|
|
def test_calc_distance_left_vector_invalid_type(self, get_invalid_vector_dict):
|
|
"""
|
|
target: test calculated distance with invalid vectors
|
|
method: input invalid vectors type
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
invalid_vector = get_invalid_vector_dict
|
|
if not isinstance(invalid_vector, dict):
|
|
self.utility_wrap.calc_distance(invalid_vector, invalid_vector,
|
|
check_task=CheckTasks.err_res,
|
|
check_items={"err_code": 1,
|
|
"err_msg": "vectors_left value {} "
|
|
"is illegal".format(invalid_vector)})
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
@pytest.mark.skip(reason="calc_distance interface is no longer supported")
|
|
def test_calc_distance_left_vector_invalid_value(self, get_invalid_vector_dict):
|
|
"""
|
|
target: test calculated distance with invalid vectors
|
|
method: input invalid vectors value
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
invalid_vector = get_invalid_vector_dict
|
|
if isinstance(invalid_vector, dict):
|
|
self.utility_wrap.calc_distance(invalid_vector, invalid_vector,
|
|
check_task=CheckTasks.err_res,
|
|
check_items={"err_code": 1,
|
|
"err_msg": "vectors_left value {} "
|
|
"is illegal".format(invalid_vector)})
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
@pytest.mark.skip(reason="calc_distance interface is no longer supported")
|
|
def test_calc_distance_right_vector_invalid_type(self, get_invalid_vector_dict):
|
|
"""
|
|
target: test calculated distance with invalid vectors
|
|
method: input invalid vectors type
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
invalid_vector = get_invalid_vector_dict
|
|
vector = cf.gen_vectors(default_nb, default_dim)
|
|
op_l = {"float_vectors": vector}
|
|
if not isinstance(invalid_vector, dict):
|
|
self.utility_wrap.calc_distance(op_l, invalid_vector,
|
|
check_task=CheckTasks.err_res,
|
|
check_items={"err_code": 1,
|
|
"err_msg": "vectors_right value {} "
|
|
"is illegal".format(invalid_vector)})
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
@pytest.mark.skip(reason="calc_distance interface is no longer supported")
|
|
def test_calc_distance_right_vector_invalid_value(self, get_invalid_vector_dict):
|
|
"""
|
|
target: test calculated distance with invalid vectors
|
|
method: input invalid vectors value
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
invalid_vector = get_invalid_vector_dict
|
|
vector = cf.gen_vectors(default_nb, default_dim)
|
|
op_l = {"float_vectors": vector}
|
|
if isinstance(invalid_vector, dict):
|
|
self.utility_wrap.calc_distance(op_l, invalid_vector,
|
|
check_task=CheckTasks.err_res,
|
|
check_items={"err_code": 1,
|
|
"err_msg": "vectors_right value {} "
|
|
"is illegal".format(invalid_vector)})
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_rename_collection_old_invalid_type(self, get_invalid_type_collection_name):
|
|
"""
|
|
target: test rename_collection when the type of old collection name is not valid
|
|
method: input not invalid collection name
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
collection_w, vectors, _, insert_ids, _ = self.init_collection_general(prefix)
|
|
old_collection_name = get_invalid_type_collection_name
|
|
new_collection_name = cf.gen_unique_str(prefix)
|
|
self.utility_wrap.rename_collection(old_collection_name, new_collection_name,
|
|
check_task=CheckTasks.err_res,
|
|
check_items={"err_code": 999,
|
|
"err_msg": "`collection_name` value {} is illegal".format(
|
|
old_collection_name)})
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_rename_collection_old_invalid_value(self, get_invalid_value_collection_name):
|
|
"""
|
|
target: test rename_collection when the value of old collection name is not valid
|
|
method: input not invalid collection name
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
collection_w, vectors, _, insert_ids, _ = self.init_collection_general(prefix)
|
|
old_collection_name = get_invalid_value_collection_name
|
|
new_collection_name = cf.gen_unique_str(prefix)
|
|
error = {"err_code": 4, "err_msg": "collection not found"}
|
|
if old_collection_name in [None, ""]:
|
|
error = {"err_code": 999, "err_msg": "is illegal"}
|
|
self.utility_wrap.rename_collection(old_collection_name, new_collection_name,
|
|
check_task=CheckTasks.err_res,
|
|
check_items=error)
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_rename_collection_new_invalid_type(self, get_invalid_type_collection_name):
|
|
"""
|
|
target: test rename_collection when the type of new collection name is not valid
|
|
method: input not invalid collection name
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
collection_w, vectors, _, insert_ids, _ = self.init_collection_general(prefix)
|
|
old_collection_name = collection_w.name
|
|
new_collection_name = get_invalid_type_collection_name
|
|
self.utility_wrap.rename_collection(old_collection_name, new_collection_name,
|
|
check_task=CheckTasks.err_res,
|
|
check_items={"err_code": 1,
|
|
"err_msg": "`collection_name` value {} is "
|
|
"illegal".format(new_collection_name)})
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_rename_collection_new_invalid_value(self, get_invalid_value_collection_name):
|
|
"""
|
|
target: test rename_collection when the value of new collection name is not valid
|
|
method: input not invalid collection name
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
collection_w, vectors, _, insert_ids, _ = self.init_collection_general(prefix)
|
|
old_collection_name = collection_w.name
|
|
new_collection_name = get_invalid_value_collection_name
|
|
error = {"err_code": 1100, "err_msg": "Invalid collection name"}
|
|
if new_collection_name in [None, ""]:
|
|
error = {"err_code": 999, "err_msg": f"`collection_name` value {new_collection_name} is illegal"}
|
|
self.utility_wrap.rename_collection(old_collection_name, new_collection_name,
|
|
check_task=CheckTasks.err_res, check_items=error)
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_rename_collection_not_existed_collection(self):
|
|
"""
|
|
target: test rename_collection when the collection name does not exist
|
|
method: input not existing collection name
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
collection_w, vectors, _, insert_ids, _ = self.init_collection_general(prefix)
|
|
old_collection_name = "test_collection_non_exist"
|
|
new_collection_name = cf.gen_unique_str(prefix)
|
|
self.utility_wrap.rename_collection(old_collection_name, new_collection_name,
|
|
check_task=CheckTasks.err_res,
|
|
check_items={"err_code": 100,
|
|
"err_msg": "collection not found"})
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_rename_collection_existed_collection_name(self):
|
|
"""
|
|
target: test rename_collection when the collection name is existed
|
|
method: input existing collection name
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
collection_w, vectors, _, insert_ids, _ = self.init_collection_general(prefix)
|
|
old_collection_name = collection_w.name
|
|
self.utility_wrap.rename_collection(old_collection_name, old_collection_name,
|
|
check_task=CheckTasks.err_res,
|
|
check_items={"err_code": 1100,
|
|
"err_msg": "collection name or database name should be different"})
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_rename_collection_existed_collection_alias(self):
|
|
"""
|
|
target: test rename_collection when the collection alias is existed
|
|
method: input existing collection alias
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
collection_w, vectors, _, insert_ids, _ = self.init_collection_general(prefix)
|
|
old_collection_name = collection_w.name
|
|
alias = "test_alias"
|
|
self.utility_wrap.create_alias(old_collection_name, alias)
|
|
self.utility_wrap.rename_collection(old_collection_name, alias,
|
|
check_task=CheckTasks.err_res,
|
|
check_items={"err_code": 1601,
|
|
"err_msg": f"alias and collection name conflict[database=default][alias={alias}]"})
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_rename_collection_using_alias(self):
|
|
"""
|
|
target: test rename_collection using alias
|
|
method: rename collection using alias name
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
collection_w = self.init_collection_general(prefix)[0]
|
|
old_collection_name = collection_w.name
|
|
alias = cf.gen_unique_str(prefix + "alias")
|
|
new_collection_name = cf.gen_unique_str(prefix + "new")
|
|
self.utility_wrap.create_alias(old_collection_name, alias)
|
|
self.utility_wrap.rename_collection(alias, new_collection_name,
|
|
check_task=CheckTasks.err_res,
|
|
check_items={"err_code": 1,
|
|
"err_msg": "unsupported use an alias to "
|
|
"rename collection, alias:{}".format(alias)})
|
|
|
|
|
|
class TestUtilityBase(TestcaseBase):
|
|
""" Test case of index interface """
|
|
|
|
@pytest.fixture(scope="function", params=["metric_type", "metric"])
|
|
def metric_field(self, request):
|
|
yield request.param
|
|
|
|
@pytest.fixture(scope="function", params=[True, False])
|
|
def sqrt(self, request):
|
|
yield request.param
|
|
|
|
@pytest.fixture(scope="function", params=["L2", "IP"])
|
|
def metric(self, request):
|
|
yield request.param
|
|
|
|
@pytest.fixture(scope="function", params=["HAMMING", "JACCARD"])
|
|
def metric_binary(self, request):
|
|
yield request.param
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_has_collection(self):
|
|
"""
|
|
target: test has_collection with collection name
|
|
method: input collection name created before
|
|
expected: True
|
|
"""
|
|
cw = self.init_collection_wrap()
|
|
res, _ = self.utility_wrap.has_collection(cw.name)
|
|
assert res is True
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_has_collection_not_created(self):
|
|
"""
|
|
target: test has_collection with collection name which is not created
|
|
method: input random collection name
|
|
expected: False
|
|
"""
|
|
c_name = cf.gen_unique_str(prefix)
|
|
_ = self.init_collection_wrap()
|
|
res, _ = self.utility_wrap.has_collection(c_name)
|
|
assert res is False
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_has_collection_after_drop(self):
|
|
"""
|
|
target: test has_collection with collection name droped before
|
|
method: input random collection name
|
|
expected: False
|
|
"""
|
|
c_name = cf.gen_unique_str(prefix)
|
|
cw = self.init_collection_wrap(name=c_name)
|
|
res, _ = self.utility_wrap.has_collection(c_name)
|
|
assert res is True
|
|
cw.drop()
|
|
res, _ = self.utility_wrap.has_collection(c_name)
|
|
assert res is False
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_has_partition(self):
|
|
"""
|
|
target: test has_partition with partition name
|
|
method: input collection name and partition name created before
|
|
expected: True
|
|
"""
|
|
c_name = cf.gen_unique_str(prefix)
|
|
p_name = cf.gen_unique_str(prefix)
|
|
cw = self.init_collection_wrap(name=c_name)
|
|
self.init_partition_wrap(cw, p_name)
|
|
res, _ = self.utility_wrap.has_partition(c_name, p_name)
|
|
assert res is True
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_has_partition_not_created(self):
|
|
"""
|
|
target: test has_partition with partition name
|
|
method: input collection name, and partition name not created before
|
|
expected: True
|
|
"""
|
|
c_name = cf.gen_unique_str(prefix)
|
|
p_name = cf.gen_unique_str()
|
|
self.init_collection_wrap(name=c_name)
|
|
res, _ = self.utility_wrap.has_partition(c_name, p_name)
|
|
assert res is False
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_has_partition_after_drop(self):
|
|
"""
|
|
target: test has_partition with partition name
|
|
method: input collection name, and partition name dropped
|
|
expected: True
|
|
"""
|
|
c_name = cf.gen_unique_str(prefix)
|
|
p_name = cf.gen_unique_str()
|
|
cw = self.init_collection_wrap(name=c_name)
|
|
pw = self.init_partition_wrap(cw, p_name)
|
|
res, _ = self.utility_wrap.has_partition(c_name, p_name)
|
|
assert res is True
|
|
pw.drop()
|
|
res, _ = self.utility_wrap.has_partition(c_name, p_name)
|
|
assert res is False
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_has_default_partition(self):
|
|
"""
|
|
target: test has_partition with '_default' partition
|
|
method: input collection name and partition name created before
|
|
expected: True
|
|
"""
|
|
c_name = cf.gen_unique_str(prefix)
|
|
self.init_collection_wrap(name=c_name)
|
|
res, _ = self.utility_wrap.has_partition(c_name, ct.default_partition_name)
|
|
assert res is True
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_list_collections(self):
|
|
"""
|
|
target: test list_collections
|
|
method: create collection, list_collections
|
|
expected: in the result
|
|
"""
|
|
c_name = cf.gen_unique_str(prefix)
|
|
self.init_collection_wrap(name=c_name)
|
|
res, _ = self.utility_wrap.list_collections()
|
|
assert c_name in res
|
|
|
|
# TODO: make sure all collections deleted
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def _test_list_collections_no_collection(self):
|
|
"""
|
|
target: test list_collections
|
|
method: no collection created, list_collections
|
|
expected: length of the result equals to 0
|
|
"""
|
|
self._connect()
|
|
res, _ = self.utility_wrap.list_collections()
|
|
assert len(res) == 0
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_index_process_collection_not_existed(self):
|
|
"""
|
|
target: test building_process
|
|
method: input collection not created before
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
c_name = cf.gen_unique_str(prefix)
|
|
self.utility_wrap.index_building_progress(
|
|
c_name,
|
|
check_task=CheckTasks.err_res,
|
|
check_items={ct.err_code: 4, ct.err_msg: "collection not found"})
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_index_process_collection_empty(self):
|
|
"""
|
|
target: test building_process
|
|
method: input empty collection
|
|
expected: no exception raised
|
|
"""
|
|
c_name = cf.gen_unique_str(prefix)
|
|
cw = self.init_collection_wrap(name=c_name)
|
|
self.index_wrap.init_index(cw.collection, default_field_name, default_index_params)
|
|
res, _ = self.utility_wrap.index_building_progress(c_name)
|
|
exp_res = {'total_rows': 0, 'indexed_rows': 0, 'pending_index_rows': 0, 'state': 'Finished'}
|
|
assert res == exp_res
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_index_process_collection_insert_no_index(self):
|
|
"""
|
|
target: test building_process
|
|
method: insert 1 entity, no index created
|
|
expected: no exception raised
|
|
"""
|
|
nb = 1
|
|
c_name = cf.gen_unique_str(prefix)
|
|
cw = self.init_collection_wrap(name=c_name)
|
|
data = cf.gen_default_list_data(nb)
|
|
cw.insert(data=data)
|
|
error = {ct.err_code: 999, ct.err_msg: f"index not found[collection={c_name}]"}
|
|
self.utility_wrap.index_building_progress(c_name, check_task=CheckTasks.err_res, check_items=error)
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_index_process_collection_index(self):
|
|
"""
|
|
target: test building_process
|
|
method: 1.insert 1024 (because minSegmentSizeToEnableIndex=1024)
|
|
2.build(server does create index) and call building_process
|
|
expected: indexed_rows=nb
|
|
"""
|
|
nb = 1024
|
|
c_name = cf.gen_unique_str(prefix)
|
|
cw = self.init_collection_wrap(name=c_name)
|
|
data = cf.gen_default_list_data(nb)
|
|
cw.insert(data=data)
|
|
cw.create_index(default_field_name, default_index_params)
|
|
cw.flush()
|
|
res, _ = self.utility_wrap.index_building_progress(c_name)
|
|
# The indexed_rows may be 0 due to compaction,
|
|
# remove this assertion for now
|
|
# assert res['indexed_rows'] == nb
|
|
assert res['total_rows'] == nb
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
@pytest.mark.skip(reason='wait to modify')
|
|
def test_index_process_collection_indexing(self):
|
|
"""
|
|
target: test building_process
|
|
method: 1.insert 2048 entities to ensure that server will build
|
|
2.call building_process during building
|
|
expected: 2048 or less entities indexed
|
|
"""
|
|
nb = 2048
|
|
c_name = cf.gen_unique_str(prefix)
|
|
cw = self.init_collection_wrap(name=c_name)
|
|
data = cf.gen_default_list_data(nb)
|
|
cw.insert(data=data)
|
|
cw.create_index(default_field_name, default_index_params)
|
|
cw.flush()
|
|
start = time.time()
|
|
while True:
|
|
time.sleep(1)
|
|
res, _ = self.utility_wrap.index_building_progress(c_name)
|
|
if 0 < res['indexed_rows'] <= nb:
|
|
break
|
|
if time.time() - start > 5:
|
|
raise MilvusException(1, f"Index build completed in more than 5s")
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_wait_index_collection_not_existed(self):
|
|
"""
|
|
target: test wait_index
|
|
method: input collection not created before
|
|
expected: raise exception
|
|
"""
|
|
self._connect()
|
|
c_name = cf.gen_unique_str(prefix)
|
|
self.utility_wrap.wait_for_index_building_complete(
|
|
c_name,
|
|
check_task=CheckTasks.err_res,
|
|
check_items={ct.err_code: 4, ct.err_msg: "collection not found"})
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_wait_index_collection_empty(self):
|
|
"""
|
|
target: test wait_index
|
|
method: input empty collection
|
|
expected: no exception raised
|
|
"""
|
|
self._connect()
|
|
c_name = cf.gen_unique_str(prefix)
|
|
cw = self.init_collection_wrap(name=c_name)
|
|
cw.create_index(default_field_name, default_index_params)
|
|
assert self.utility_wrap.wait_for_index_building_complete(c_name)[0]
|
|
res, _ = self.utility_wrap.index_building_progress(c_name)
|
|
exp_res = {'total_rows': 0, 'indexed_rows': 0, 'pending_index_rows': 0, 'state': 'Finished'}
|
|
assert res == exp_res
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_wait_index_collection_index(self):
|
|
"""
|
|
target: test wait_index
|
|
method: insert 5000 entities, build and call wait_index
|
|
expected: 5000 entity indexed
|
|
"""
|
|
nb = 5000
|
|
c_name = cf.gen_unique_str(prefix)
|
|
cw = self.init_collection_wrap(name=c_name)
|
|
data = cf.gen_default_list_data(nb)
|
|
cw.insert(data=data)
|
|
cw.flush()
|
|
cw.create_index(default_field_name, default_index_params)
|
|
res, _ = self.utility_wrap.wait_for_index_building_complete(c_name)
|
|
assert res is True
|
|
res, _ = self.utility_wrap.index_building_progress(c_name)
|
|
assert res["indexed_rows"] == nb
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_loading_progress_without_loading(self):
|
|
"""
|
|
target: test loading progress without loading
|
|
method: insert and flush data, call loading_progress without loading
|
|
expected: raise exception
|
|
"""
|
|
collection_w = self.init_collection_wrap()
|
|
df = cf.gen_default_dataframe_data()
|
|
collection_w.insert(df)
|
|
assert collection_w.num_entities == ct.default_nb
|
|
self.utility_wrap.loading_progress(collection_w.name,
|
|
check_task=CheckTasks.err_res,
|
|
check_items={ct.err_code: 101,
|
|
ct.err_msg: 'collection not loaded'})
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
@pytest.mark.parametrize("nb", [ct.default_nb, 5000])
|
|
def test_loading_progress_collection(self, nb):
|
|
"""
|
|
target: test loading progress
|
|
method: 1.insert flush and load 2.call loading_progress
|
|
expected: all entities is loafed, because load is synchronous
|
|
"""
|
|
# create, insert default_nb, flush and load
|
|
collection_w = self.init_collection_general(prefix, insert_data=True, nb=nb)[0]
|
|
res, _ = self.utility_wrap.loading_progress(collection_w.name)
|
|
assert res[loading_progress] == '100%'
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_loading_progress_with_async_load(self):
|
|
"""
|
|
target: test loading progress with async collection load
|
|
method: 1.load collection with async=True 2.loading_progress
|
|
expected: loading part entities
|
|
"""
|
|
collection_w = self.init_collection_wrap()
|
|
df = cf.gen_default_dataframe_data()
|
|
collection_w.insert(df)
|
|
assert collection_w.num_entities == ct.default_nb
|
|
collection_w.create_index(ct.default_float_vec_field_name, index_params=ct.default_flat_index)
|
|
collection_w.load(_async=True)
|
|
res, _ = self.utility_wrap.loading_progress(collection_w.name)
|
|
loading_int = cf.percent_to_int(res[loading_progress])
|
|
if -1 != loading_int:
|
|
assert (0 <= loading_int <= 100)
|
|
else:
|
|
log.info("The output of loading progress is not a string or a percentage")
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_loading_progress_empty_collection(self):
|
|
"""
|
|
target: test loading_progress on an empty collection
|
|
method: 1.create collection and no insert 2.loading_progress
|
|
expected: 0 entities is loaded
|
|
"""
|
|
collection_w = self.init_collection_wrap()
|
|
collection_w.create_index(ct.default_float_vec_field_name, index_params=ct.default_flat_index)
|
|
collection_w.load()
|
|
res, _ = self.utility_wrap.loading_progress(collection_w.name)
|
|
exp_res = {loading_progress: '100%'}
|
|
|
|
assert exp_res == res
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_loading_progress_after_release(self):
|
|
"""
|
|
target: test loading progress after release
|
|
method: insert and flush data, call loading_progress after release
|
|
expected: return successfully with 0%
|
|
"""
|
|
collection_w = self.init_collection_general(prefix, insert_data=True, nb=100)[0]
|
|
collection_w.release()
|
|
error = {ct.err_code: 999, ct.err_msg: "collection not loaded"}
|
|
self.utility_wrap.loading_progress(collection_w.name,
|
|
check_task=CheckTasks.err_res, check_items=error)
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_loading_progress_with_release_partition(self):
|
|
"""
|
|
target: test loading progress after release part partitions
|
|
method: 1.insert data into two partitions and flush
|
|
2.load one partition and release one partition
|
|
expected: loaded one partition entities
|
|
"""
|
|
half = ct.default_nb
|
|
# insert entities into two partitions, collection flush and load
|
|
collection_w, partition_w, _, _ = self.insert_entities_into_two_partitions_in_half(half)
|
|
partition_w.release()
|
|
res = self.utility_wrap.loading_progress(collection_w.name)[0]
|
|
assert res[loading_progress] == '100%'
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_loading_progress_with_load_partition(self):
|
|
"""
|
|
target: test loading progress after load partition
|
|
method: 1.insert data into two partitions and flush
|
|
2.load one partition and loading progress
|
|
expected: loaded one partition entities
|
|
"""
|
|
half = ct.default_nb
|
|
collection_w, partition_w, _, _ = self.insert_entities_into_two_partitions_in_half(half)
|
|
collection_w.release()
|
|
partition_w.load()
|
|
res = self.utility_wrap.loading_progress(collection_w.name)[0]
|
|
assert res[loading_progress] == '100%'
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_loading_progress_with_partition(self):
|
|
"""
|
|
target: test loading progress with partition
|
|
method: 1.insert data into two partitions and flush, and load
|
|
2.loading progress with one partition
|
|
expected: loaded one partition entities
|
|
"""
|
|
half = ct.default_nb
|
|
collection_w, partition_w, _, _ = self.insert_entities_into_two_partitions_in_half(half)
|
|
res = self.utility_wrap.loading_progress(collection_w.name, partition_names=[partition_w.name])[0]
|
|
assert res[loading_progress] == '100%'
|
|
|
|
@pytest.mark.tags(CaseLabel.ClusterOnly)
|
|
def test_loading_progress_multi_replicas(self):
|
|
"""
|
|
target: test loading progress with multi replicas
|
|
method: 1.Create collection and insert data
|
|
2.Load replicas and get loading progress
|
|
3.Create partitions and insert data
|
|
4.Get loading progress
|
|
5.Release and re-load replicas, get loading progress
|
|
expected: Verify loading progress result
|
|
"""
|
|
collection_w = self.init_collection_wrap()
|
|
collection_w.insert(cf.gen_default_dataframe_data())
|
|
assert collection_w.num_entities == ct.default_nb
|
|
default_index = {"index_type": "IVF_FLAT", "params": {"nlist": 128}, "metric_type": "L2"}
|
|
collection_w.create_index("float_vector", default_index)
|
|
collection_w.load(partition_names=[ct.default_partition_name], replica_number=2)
|
|
res_collection, _ = self.utility_wrap.loading_progress(collection_w.name)
|
|
assert res_collection == {loading_progress: '100%'}
|
|
|
|
# create partition and insert
|
|
partition_w = self.init_partition_wrap(collection_wrap=collection_w)
|
|
partition_w.insert(cf.gen_default_dataframe_data(start=ct.default_nb))
|
|
assert partition_w.num_entities == ct.default_nb
|
|
res_part_partition, _ = self.utility_wrap.loading_progress(collection_w.name)
|
|
assert res_part_partition == {'loading_progress': '100%'}
|
|
|
|
res_part_partition, _ = self.utility_wrap.loading_progress(collection_w.name)
|
|
assert res_part_partition == {'loading_progress': '100%'}
|
|
|
|
collection_w.release()
|
|
collection_w.load(replica_number=2)
|
|
res_all_partitions, _ = self.utility_wrap.loading_progress(collection_w.name)
|
|
assert res_all_partitions == {'loading_progress': '100%'}
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_wait_loading_collection_empty(self):
|
|
"""
|
|
target: test wait_for_loading
|
|
method: input empty collection
|
|
expected: no exception raised
|
|
"""
|
|
self._connect()
|
|
cw = self.init_collection_wrap(name=cf.gen_unique_str(prefix))
|
|
cw.create_index(ct.default_float_vec_field_name, index_params=ct.default_flat_index)
|
|
cw.load()
|
|
self.utility_wrap.wait_for_loading_complete(cw.name)
|
|
res, _ = self.utility_wrap.loading_progress(cw.name)
|
|
exp_res = {loading_progress: "100%"}
|
|
assert res == exp_res
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_wait_for_loading_complete(self):
|
|
"""
|
|
target: test wait for loading collection
|
|
method: insert 10000 entities and wait for loading complete
|
|
expected: after loading complete, loaded entities is 10000
|
|
"""
|
|
nb = 6000
|
|
collection_w = self.init_collection_wrap()
|
|
df = cf.gen_default_dataframe_data(nb)
|
|
collection_w.insert(df, timeout=60)
|
|
assert collection_w.num_entities == nb
|
|
collection_w.create_index(ct.default_float_vec_field_name, index_params=ct.default_flat_index)
|
|
collection_w.load(_async=True)
|
|
self.utility_wrap.wait_for_loading_complete(collection_w.name, timeout=45)
|
|
res, _ = self.utility_wrap.loading_progress(collection_w.name)
|
|
assert res[loading_progress] == '100%'
|
|
|
|
@pytest.mark.tags(CaseLabel.L0)
|
|
def test_drop_collection(self):
|
|
"""
|
|
target: test utility drop collection by name
|
|
method: input collection name and drop collection
|
|
expected: collection is dropped
|
|
"""
|
|
c_name = cf.gen_unique_str(prefix)
|
|
self.init_collection_wrap(c_name)
|
|
assert self.utility_wrap.has_collection(c_name)[0]
|
|
self.utility_wrap.drop_collection(c_name)
|
|
assert not self.utility_wrap.has_collection(c_name)[0]
|
|
|
|
@pytest.mark.tags(CaseLabel.L0)
|
|
def test_drop_collection_repeatedly(self):
|
|
"""
|
|
target: test drop collection repeatedly
|
|
method: 1.collection.drop 2.utility.drop_collection
|
|
expected: raise exception
|
|
"""
|
|
c_name = cf.gen_unique_str(prefix)
|
|
collection_w = self.init_collection_wrap(c_name)
|
|
assert self.utility_wrap.has_collection(c_name)[0]
|
|
collection_w.drop()
|
|
assert not self.utility_wrap.has_collection(c_name)[0]
|
|
|
|
# error = {ct.err_code: 1, ct.err_msg: {"describe collection failed: can't find collection:"}}
|
|
# self.utility_wrap.drop_collection(c_name, check_task=CheckTasks.err_res, check_items=error)
|
|
# @longjiquan: dropping collection should be idempotent.
|
|
self.utility_wrap.drop_collection(c_name)
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_drop_collection_create_repeatedly(self):
|
|
"""
|
|
target: test repeatedly create and drop same name collection
|
|
method: repeatedly create and drop collection
|
|
expected: no exception
|
|
"""
|
|
from time import sleep
|
|
loops = 3
|
|
c_name = cf.gen_unique_str(prefix)
|
|
for _ in range(loops):
|
|
self.init_collection_wrap(c_name)
|
|
assert self.utility_wrap.has_collection(c_name)[0]
|
|
self.utility_wrap.drop_collection(c_name)
|
|
assert not self.utility_wrap.has_collection(c_name)[0]
|
|
sleep(1)
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_rename_collection(self):
|
|
"""
|
|
target: test rename collection function to single collection
|
|
method: call rename_collection API to rename collection
|
|
expected: collection renamed successfully without any change on aliases
|
|
"""
|
|
self._connect()
|
|
collection_w, vectors, _, insert_ids, _ = self.init_collection_general(prefix)
|
|
old_collection_name = collection_w.name
|
|
new_collection_name = cf.gen_unique_str(prefix + "new")
|
|
alias = cf.gen_unique_str(prefix + "alias")
|
|
self.utility_wrap.create_alias(old_collection_name, alias)
|
|
self.utility_wrap.rename_collection(old_collection_name, new_collection_name)
|
|
collection_w = self.init_collection_wrap(name=new_collection_name,
|
|
check_task=CheckTasks.check_collection_property,
|
|
check_items={exp_name: new_collection_name,
|
|
exp_schema: default_schema})
|
|
collections = self.utility_wrap.list_collections()[0]
|
|
assert new_collection_name in collections
|
|
assert old_collection_name not in collections
|
|
assert [alias] == collection_w.aliases
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_rename_collections(self):
|
|
"""
|
|
target: test rename collection function to multiple collections
|
|
method: call rename_collection API to rename collections
|
|
expected: collections renamed successfully without any change on aliases
|
|
"""
|
|
self._connect()
|
|
# create two collections
|
|
collection_w_1 = self.init_collection_general(prefix)[0]
|
|
old_collection_name_1 = collection_w_1.name
|
|
collection_w_2 = self.init_collection_general(prefix)[0]
|
|
old_collection_name_2 = collection_w_2.name
|
|
tmp_collection_name = cf.gen_unique_str(prefix + "tmp")
|
|
# create alias to each collection
|
|
alias_1 = cf.gen_unique_str(prefix + "alias1")
|
|
alias_2 = cf.gen_unique_str(prefix + "alias2")
|
|
self.utility_wrap.create_alias(old_collection_name_1, alias_1)
|
|
self.utility_wrap.create_alias(old_collection_name_2, alias_2)
|
|
# switch name of the existing collections
|
|
self.utility_wrap.rename_collection(old_collection_name_1, tmp_collection_name)
|
|
self.utility_wrap.rename_collection(old_collection_name_2, old_collection_name_1)
|
|
self.utility_wrap.rename_collection(tmp_collection_name, old_collection_name_2)
|
|
# check collection renamed successfully
|
|
collection_w_1 = self.init_collection_wrap(name=old_collection_name_1,
|
|
check_task=CheckTasks.check_collection_property,
|
|
check_items={exp_name: old_collection_name_1,
|
|
exp_schema: default_schema})
|
|
collection_w_2 = self.init_collection_wrap(name=old_collection_name_2,
|
|
check_task=CheckTasks.check_collection_property,
|
|
check_items={exp_name: old_collection_name_2,
|
|
exp_schema: default_schema})
|
|
collections = self.utility_wrap.list_collections()[0]
|
|
assert old_collection_name_1 in collections
|
|
assert old_collection_name_2 in collections
|
|
assert tmp_collection_name not in collections
|
|
# check alias not changed
|
|
assert collection_w_1.aliases[0] == alias_2
|
|
assert collection_w_2.aliases[0] == alias_1
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
@pytest.mark.skip("move to test_milvus_client_rename_back_old_collection")
|
|
def test_rename_back_old_collection(self):
|
|
"""
|
|
target: test rename collection function to single collection
|
|
method: rename back to old collection name
|
|
expected: collection renamed successfully without any change on aliases
|
|
"""
|
|
# 1. connect
|
|
self._connect()
|
|
# 2. create a collection
|
|
collection_w, vectors, _, insert_ids, _ = self.init_collection_general(prefix)
|
|
old_collection_name = collection_w.name
|
|
new_collection_name = cf.gen_unique_str(prefix + "new")
|
|
alias = cf.gen_unique_str(prefix + "alias")
|
|
# 3. create an alias
|
|
self.utility_wrap.create_alias(old_collection_name, alias)
|
|
collection_alias = collection_w.aliases
|
|
# 4. rename collection
|
|
self.utility_wrap.rename_collection(old_collection_name, new_collection_name)
|
|
# 5. rename back to old collection name
|
|
self.utility_wrap.rename_collection(new_collection_name, old_collection_name)
|
|
collection_w = self.init_collection_wrap(name=old_collection_name,
|
|
check_task=CheckTasks.check_collection_property,
|
|
check_items={exp_name: old_collection_name,
|
|
exp_schema: default_schema})
|
|
collections = self.utility_wrap.list_collections()[0]
|
|
assert old_collection_name in collections
|
|
assert new_collection_name not in collections
|
|
assert collection_alias == collection_w.aliases
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
@pytest.mark.skip("move to test_milvus_client_rename_back_old_alias")
|
|
def test_rename_back_old_alias(self):
|
|
"""
|
|
target: test rename collection function to single collection
|
|
method: rename back to old collection alias
|
|
expected: collection renamed successfully without any change on aliases
|
|
"""
|
|
# 1. connect
|
|
self._connect()
|
|
# 2. create a collection
|
|
collection_w, vectors, _, insert_ids, _ = self.init_collection_general(prefix)
|
|
old_collection_name = collection_w.name
|
|
alias = cf.gen_unique_str(prefix + "alias")
|
|
# 3. create an alias
|
|
self.utility_wrap.create_alias(old_collection_name, alias)
|
|
collection_alias = collection_w.aliases
|
|
log.info(collection_alias)
|
|
# 4. drop the alias
|
|
self.utility_wrap.drop_alias(collection_alias[0])
|
|
# 5. rename collection to the dropped alias name
|
|
self.utility_wrap.rename_collection(old_collection_name, collection_alias[0])
|
|
self.init_collection_wrap(name=collection_alias[0],
|
|
check_task=CheckTasks.check_collection_property,
|
|
check_items={exp_name: collection_alias[0],
|
|
exp_schema: default_schema})
|
|
collections = self.utility_wrap.list_collections()[0]
|
|
assert collection_alias[0] in collections
|
|
assert old_collection_name not in collections
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_create_alias_using_dropped_collection_name(self):
|
|
"""
|
|
target: test create alias using a dropped collection name
|
|
method: create 2 collections and drop one collection
|
|
expected: raise no exception
|
|
"""
|
|
# 1. create 2 collections
|
|
a_name = cf.gen_unique_str("aa")
|
|
b_name = cf.gen_unique_str("bb")
|
|
self.init_collection_wrap(name=a_name, schema=default_schema,
|
|
check_task=CheckTasks.check_collection_property,
|
|
check_items={exp_name: a_name, exp_schema: default_schema})
|
|
self.init_collection_wrap(name=b_name, schema=default_schema,
|
|
check_task=CheckTasks.check_collection_property,
|
|
check_items={exp_name: b_name, exp_schema: default_schema})
|
|
|
|
# 2. drop collection a
|
|
self.utility_wrap.drop_collection(a_name)
|
|
assert self.utility_wrap.has_collection(a_name)[0] is False
|
|
assert len(self.utility_wrap.list_aliases(b_name)[0]) == 0
|
|
|
|
# 3. create alias with the name of collection a
|
|
self.utility_wrap.create_alias(b_name, a_name)
|
|
b_alias, _ = self.utility_wrap.list_aliases(b_name)
|
|
assert a_name in b_alias
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_list_indexes(self):
|
|
"""
|
|
target: test utility.list_indexes
|
|
method: create 2 collections and list indexes
|
|
expected: raise no exception
|
|
"""
|
|
# 1. create 2 collections
|
|
string_field = ct.default_string_field_name
|
|
collection_w1 = self.init_collection_general(prefix, True)[0]
|
|
collection_w2 = self.init_collection_general(prefix, True, is_index=False)[0]
|
|
collection_w2.create_index(string_field)
|
|
|
|
# 2. list indexes
|
|
res1, _ = self.utility_wrap.list_indexes(collection_w1.name)
|
|
assert res1 == [ct.default_float_vec_field_name]
|
|
res2, _ = self.utility_wrap.list_indexes(collection_w2.name)
|
|
assert res2 == [string_field]
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_get_server_type(self):
|
|
"""
|
|
target: test utility.get_server_type
|
|
method: get_server_type
|
|
expected: raise no exception
|
|
"""
|
|
self._connect()
|
|
res, _ = self.utility_wrap.get_server_type()
|
|
assert res == "milvus"
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_load_state(self):
|
|
"""
|
|
target: test utility.load_state
|
|
method: load_state
|
|
expected: raise no exception
|
|
"""
|
|
collection_w = self.init_collection_general(prefix, True, partition_num=1)[0]
|
|
res1, _ = self.utility_wrap.load_state(collection_w.name)
|
|
assert str(res1) == "Loaded"
|
|
collection_w.release()
|
|
res2, _ = self.utility_wrap.load_state(collection_w.name)
|
|
assert str(res2) == "NotLoad"
|
|
collection_w.load(partition_names=[ct.default_partition_name])
|
|
res3, _ = self.utility_wrap.load_state(collection_w.name)
|
|
assert str(res3) == "Loaded"
|
|
|
|
|
|
class TestUtilityAdvanced(TestcaseBase):
|
|
""" Test case of index interface """
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_has_collection_multi_collections(self):
|
|
"""
|
|
target: test has_collection with collection name
|
|
method: input collection name created before
|
|
expected: True
|
|
"""
|
|
c_name = cf.gen_unique_str(prefix)
|
|
c_name_2 = cf.gen_unique_str(prefix)
|
|
self.init_collection_wrap(name=c_name)
|
|
self.init_collection_wrap(name=c_name_2)
|
|
for name in [c_name, c_name_2]:
|
|
res, _ = self.utility_wrap.has_collection(name)
|
|
assert res is True
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_list_collections_multi_collection(self):
|
|
"""
|
|
target: test list_collections
|
|
method: create collection, list_collections
|
|
expected: in the result
|
|
"""
|
|
c_name = cf.gen_unique_str(prefix)
|
|
c_name_2 = cf.gen_unique_str(prefix)
|
|
self.init_collection_wrap(name=c_name)
|
|
self.init_collection_wrap(name=c_name_2)
|
|
res, _ = self.utility_wrap.list_collections()
|
|
for name in [c_name, c_name_2]:
|
|
assert name in res
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_drop_multi_collection_concurrent(self):
|
|
"""
|
|
target: test concurrent drop collection
|
|
method: multi thread drop one collection
|
|
expected: drop successfully
|
|
"""
|
|
thread_num = 3
|
|
threads = []
|
|
c_names = []
|
|
num = 5
|
|
|
|
for i in range(thread_num * num):
|
|
c_name = cf.gen_unique_str(prefix)
|
|
self.init_collection_wrap(c_name)
|
|
c_names.append(c_name)
|
|
|
|
def create_and_drop_collection(names):
|
|
for name in names:
|
|
assert self.utility_wrap.has_collection(name)[0]
|
|
self.utility_wrap.drop_collection(name)
|
|
assert not self.utility_wrap.has_collection(name)[0]
|
|
|
|
for i in range(thread_num):
|
|
x = threading.Thread(target=create_and_drop_collection, args=(c_names[i * num:(i + 1) * num],))
|
|
threads.append(x)
|
|
x.start()
|
|
for t in threads:
|
|
t.join()
|
|
log.debug(self.utility_wrap.list_collections()[0])
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_get_query_segment_info_empty_collection(self):
|
|
"""
|
|
target: test getting query segment info of empty collection
|
|
method: init a collection and get query segment info
|
|
expected: length of segment is 0
|
|
"""
|
|
c_name = cf.gen_unique_str(prefix)
|
|
collection_w = self.init_collection_wrap(name=c_name)
|
|
collection_w.create_index(ct.default_float_vec_field_name, index_params=ct.default_flat_index)
|
|
collection_w.load()
|
|
res, _ = self.utility_wrap.get_query_segment_info(c_name)
|
|
assert len(res) == 0
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
@pytest.mark.parametrize("primary_field", ["int64_pk", "varchar_pk"])
|
|
def test_get_sealed_query_segment_info(self, primary_field):
|
|
"""
|
|
target: test getting sealed query segment info of collection with data
|
|
method: init a collection, insert data, flush, index, load, and get query segment info
|
|
expected:
|
|
1. length of segment is greater than 0
|
|
2. the sum num_rows of each segment is equal to num of entities
|
|
3. all segment is_sorted true
|
|
"""
|
|
nb = 3000
|
|
segment_num = 2
|
|
collection_name = cf.gen_unique_str(prefix)
|
|
|
|
# connect -> create collection
|
|
self._connect()
|
|
self.collection_wrap.init_collection(
|
|
name=collection_name,
|
|
schema=cf.set_collection_schema(
|
|
fields=[primary_field, ct.default_float_vec_field_name],
|
|
field_params={
|
|
primary_field: FieldParams(is_primary=True, max_length=128).to_dict,
|
|
ct.default_float_vec_field_name: FieldParams(dim=ct.default_dim).to_dict,
|
|
},
|
|
)
|
|
)
|
|
|
|
for _ in range(segment_num):
|
|
# insert random pks
|
|
data = cf.gen_values(self.collection_wrap.schema, nb=nb, random_pk=True)
|
|
self.collection_wrap.insert(data)
|
|
self.collection_wrap.flush()
|
|
|
|
# flush -> index -> load -> sealed segments is sorted
|
|
self.build_multi_index(index_params=DefaultVectorIndexParams.IVF_SQ8(ct.default_float_vec_field_name))
|
|
self.collection_wrap.load()
|
|
|
|
# get_query_segment_info and verify results
|
|
res_sealed, _ = self.utility_wrap.get_query_segment_info(collection_name)
|
|
assert len(res_sealed) > 0 # maybe mix compaction to 1 segment
|
|
cnt = 0
|
|
for r in res_sealed:
|
|
log.info(f"segmentID {r.segmentID}: state: {r.state}; num_rows: {r.num_rows}; is_sorted: {r.is_sorted} ")
|
|
cnt += r.num_rows
|
|
assert r.is_sorted is True
|
|
assert cnt == nb * segment_num
|
|
|
|
# verify search
|
|
self.collection_wrap.search(data=cf.gen_vectors(ct.default_nq, ct.default_dim),
|
|
anns_field=ct.default_float_vec_field_name, param=DefaultVectorSearchParams.IVF_SQ8(),
|
|
limit=ct.default_limit,
|
|
check_task=CheckTasks.check_search_results,
|
|
check_items={"nq": ct.default_nq,
|
|
"limit": ct.default_limit})
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_get_growing_query_segment_info(self):
|
|
"""
|
|
target: test getting growing query segment info of collection with data
|
|
method: init a collection, index, load, insert data, and get query segment info
|
|
expected:
|
|
1. length of segment is 0, growing segment is not visible for get_query_segment_info
|
|
"""
|
|
nb = 3000
|
|
primary_field = "int64"
|
|
collection_name = cf.gen_unique_str(prefix)
|
|
|
|
# connect -> create collection
|
|
self._connect()
|
|
self.collection_wrap.init_collection(
|
|
name=collection_name,
|
|
schema=cf.set_collection_schema(
|
|
fields=[primary_field, ct.default_float_vec_field_name],
|
|
field_params={
|
|
primary_field: FieldParams(is_primary=True, max_length=128).to_dict,
|
|
ct.default_float_vec_field_name: FieldParams(dim=ct.default_dim).to_dict,
|
|
},
|
|
)
|
|
)
|
|
|
|
# index -> load
|
|
self.build_multi_index(index_params=DefaultVectorIndexParams.IVF_SQ8(ct.default_float_vec_field_name))
|
|
self.collection_wrap.load()
|
|
|
|
# insert random pks ***
|
|
data = cf.gen_values(self.collection_wrap.schema, nb=nb, random_pk=True)
|
|
self.collection_wrap.insert(data)
|
|
|
|
# get_query_segment_info and verify results
|
|
res_sealed, _ = self.utility_wrap.get_query_segment_info(collection_name)
|
|
assert len(res_sealed) == 0
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_get_sealed_query_segment_info_after_create_index(self):
|
|
"""
|
|
target: test getting sealed query segment info of collection with data
|
|
method: init a collection, insert data, flush, create index, load, and get query segment info
|
|
expected:
|
|
1. length of segment is greater than 0
|
|
2. the sum num_rows of each segment is equal to num of entities
|
|
"""
|
|
c_name = cf.gen_unique_str(prefix)
|
|
collection_w = self.init_collection_wrap(name=c_name)
|
|
nb = 3000
|
|
df = cf.gen_default_dataframe_data(nb)
|
|
collection_w.insert(df)
|
|
collection_w.num_entities
|
|
collection_w.create_index(default_field_name, default_index_params)
|
|
collection_w.load()
|
|
res, _ = self.utility_wrap.get_query_segment_info(c_name)
|
|
assert len(res) > 0
|
|
segment_ids = []
|
|
cnt = 0
|
|
for r in res:
|
|
log.info(f"segmentID {r.segmentID}: state: {r.state}; num_rows: {r.num_rows} ")
|
|
if r.segmentID not in segment_ids:
|
|
segment_ids.append(r.segmentID)
|
|
cnt += r.num_rows
|
|
assert cnt == nb
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_load_balance_normal(self):
|
|
"""
|
|
target: test load balance of collection
|
|
method: init a collection and load balance
|
|
expected: sealed_segment_ids is subset of des_sealed_segment_ids
|
|
"""
|
|
# init a collection
|
|
self._connect()
|
|
querynode_num = len(MilvusSys().query_nodes)
|
|
if querynode_num < 2:
|
|
pytest.skip("skip load balance testcase when querynode number less than 2")
|
|
c_name = cf.gen_unique_str(prefix)
|
|
collection_w = self.init_collection_wrap(name=c_name)
|
|
collection_w.create_index(default_field_name, default_index_params)
|
|
ms = MilvusSys()
|
|
nb = 3000
|
|
df = cf.gen_default_dataframe_data(nb)
|
|
collection_w.insert(df)
|
|
# get sealed segments
|
|
collection_w.num_entities
|
|
# get growing segments
|
|
collection_w.insert(df)
|
|
collection_w.load()
|
|
# prepare load balance params
|
|
time.sleep(0.5)
|
|
res, _ = self.utility_wrap.get_query_segment_info(c_name)
|
|
segment_distribution = cf.get_segment_distribution(res)
|
|
all_querynodes = [node["identifier"] for node in ms.query_nodes]
|
|
assert len(all_querynodes) > 1
|
|
all_querynodes = sorted(all_querynodes,
|
|
key=lambda x: len(segment_distribution[x]["sealed"])
|
|
if x in segment_distribution else 0, reverse=True)
|
|
src_node_id = all_querynodes[0]
|
|
des_node_ids = all_querynodes[1:]
|
|
sealed_segment_ids = segment_distribution[src_node_id]["sealed"]
|
|
# load balance
|
|
self.utility_wrap.load_balance(collection_w.name, src_node_id, des_node_ids, sealed_segment_ids)
|
|
# get segments distribution after load balance
|
|
time.sleep(0.5)
|
|
res, _ = self.utility_wrap.get_query_segment_info(c_name)
|
|
segment_distribution = cf.get_segment_distribution(res)
|
|
sealed_segment_ids_after_load_banalce = segment_distribution[src_node_id]["sealed"]
|
|
# assert src node has no sealed segments
|
|
assert sealed_segment_ids_after_load_banalce == []
|
|
des_sealed_segment_ids = []
|
|
for des_node_id in des_node_ids:
|
|
des_sealed_segment_ids += segment_distribution[des_node_id]["sealed"]
|
|
# assert sealed_segment_ids is subset of des_sealed_segment_ids
|
|
assert set(sealed_segment_ids).issubset(des_sealed_segment_ids)
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_load_balance_with_src_node_not_exist(self):
|
|
"""
|
|
target: test load balance of collection
|
|
method: init a collection and load balance with src_node not exist
|
|
expected: raise exception
|
|
"""
|
|
# init a collection
|
|
c_name = cf.gen_unique_str(prefix)
|
|
collection_w = self.init_collection_wrap(name=c_name)
|
|
ms = MilvusSys()
|
|
nb = 3000
|
|
df = cf.gen_default_dataframe_data(nb)
|
|
collection_w.insert(df)
|
|
# get sealed segments
|
|
collection_w.num_entities
|
|
# get growing segments
|
|
collection_w.insert(df)
|
|
collection_w.create_index(ct.default_float_vec_field_name, index_params=ct.default_flat_index)
|
|
collection_w.load()
|
|
# prepare load balance params
|
|
res, _ = self.utility_wrap.get_query_segment_info(c_name)
|
|
segment_distribution = cf.get_segment_distribution(res)
|
|
all_querynodes = [node["identifier"] for node in ms.query_nodes]
|
|
all_querynodes = sorted(all_querynodes,
|
|
key=lambda x: len(segment_distribution[x]["sealed"])
|
|
if x in segment_distribution else 0, reverse=True)
|
|
# use a node id greater than all nodes (not just querynodes) to ensure it truly does not exist
|
|
all_node_ids = [node["identifier"] for node in ms.nodes]
|
|
invalid_src_node_id = max(all_node_ids) + 1
|
|
src_node_id = all_querynodes[0]
|
|
dst_node_ids = all_querynodes[1:]
|
|
sealed_segment_ids = segment_distribution[src_node_id]["sealed"]
|
|
# load balance
|
|
self.utility_wrap.load_balance(collection_w.name, invalid_src_node_id, dst_node_ids, sealed_segment_ids,
|
|
check_task=CheckTasks.err_res,
|
|
check_items={ct.err_code: 1, ct.err_msg: "source node not found in any replica"})
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_load_balance_with_all_dst_node_not_exist(self):
|
|
"""
|
|
target: test load balance of collection
|
|
method: init a collection and load balance with all dst_node not exist
|
|
expected: raise exception
|
|
"""
|
|
# init a collection
|
|
c_name = cf.gen_unique_str(prefix)
|
|
collection_w = self.init_collection_wrap(name=c_name)
|
|
ms = MilvusSys()
|
|
nb = 3000
|
|
df = cf.gen_default_dataframe_data(nb)
|
|
collection_w.insert(df)
|
|
# get sealed segments
|
|
collection_w.num_entities
|
|
# get growing segments
|
|
collection_w.insert(df)
|
|
collection_w.create_index(ct.default_float_vec_field_name, index_params=ct.default_flat_index)
|
|
collection_w.load()
|
|
# prepare load balance params
|
|
res, _ = self.utility_wrap.get_query_segment_info(c_name)
|
|
segment_distribution = cf.get_segment_distribution(res)
|
|
all_querynodes = [node["identifier"] for node in ms.query_nodes]
|
|
all_querynodes = sorted(all_querynodes,
|
|
key=lambda x: len(segment_distribution[x]["sealed"])
|
|
if x in segment_distribution else 0, reverse=True)
|
|
src_node_id = all_querynodes[0]
|
|
# use node ids greater than all nodes (not just querynodes) to ensure they truly do not exist
|
|
all_node_ids = [node["identifier"] for node in ms.nodes]
|
|
max_node_id = max(all_node_ids)
|
|
dst_node_ids = [max_node_id + 1, max_node_id + 2]
|
|
sealed_segment_ids = segment_distribution[src_node_id]["sealed"]
|
|
# load balance
|
|
self.utility_wrap.load_balance(collection_w.name, src_node_id, dst_node_ids, sealed_segment_ids,
|
|
check_task=CheckTasks.err_res,
|
|
check_items={ct.err_code: 1, ct.err_msg: "destination node not found in the same replica"})
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_load_balance_with_one_sealed_segment_id_not_exist(self):
|
|
"""
|
|
target: test load balance of collection
|
|
method: init a collection and load balance with one of sealed segment ids not exist
|
|
expected: raise exception
|
|
"""
|
|
# init a collection
|
|
c_name = cf.gen_unique_str(prefix)
|
|
collection_w = self.init_collection_wrap(name=c_name)
|
|
collection_w.create_index(default_field_name, default_index_params)
|
|
ms = MilvusSys()
|
|
nb = 3000
|
|
df = cf.gen_default_dataframe_data(nb)
|
|
collection_w.insert(df)
|
|
# get sealed segments
|
|
collection_w.flush()
|
|
# get growing segments
|
|
collection_w.insert(df)
|
|
collection_w.load()
|
|
# prepare load balance params
|
|
res, _ = self.utility_wrap.get_query_segment_info(c_name)
|
|
segment_distribution = cf.get_segment_distribution(res)
|
|
all_querynodes = [node["identifier"] for node in ms.query_nodes]
|
|
all_querynodes = sorted(all_querynodes,
|
|
key=lambda x: len(segment_distribution[x]["sealed"])
|
|
if x in segment_distribution else 0, reverse=True)
|
|
src_node_id = all_querynodes[0]
|
|
dst_node_ids = all_querynodes[1:]
|
|
sealed_segment_ids = segment_distribution[src_node_id]["sealed"]
|
|
if len(segment_distribution[src_node_id]["sealed"]) == 0:
|
|
sealed_segment_ids = [0] # add a segment id which is not exist
|
|
else:
|
|
# add a segment id which is not exist
|
|
sealed_segment_ids.append(max(segment_distribution[src_node_id]["sealed"]) + 1)
|
|
# load balance
|
|
self.utility_wrap.load_balance(collection_w.name, src_node_id, dst_node_ids, sealed_segment_ids,
|
|
check_task=CheckTasks.err_res,
|
|
check_items={ct.err_code: 999, ct.err_msg: "not found in source node"})
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
def test_load_balance_with_all_sealed_segment_id_not_exist(self):
|
|
"""
|
|
target: test load balance of collection
|
|
method: init a collection and load balance with one of sealed segment ids not exist
|
|
expected: raise exception
|
|
"""
|
|
# init a collection
|
|
c_name = cf.gen_unique_str(prefix)
|
|
collection_w = self.init_collection_wrap(name=c_name)
|
|
ms = MilvusSys()
|
|
nb = 3000
|
|
df = cf.gen_default_dataframe_data(nb)
|
|
collection_w.insert(df)
|
|
# get sealed segments
|
|
collection_w.num_entities
|
|
collection_w.create_index(default_field_name, default_index_params)
|
|
# get growing segments
|
|
collection_w.insert(df)
|
|
collection_w.load()
|
|
# prepare load balance params
|
|
res, _ = self.utility_wrap.get_query_segment_info(c_name)
|
|
segment_distribution = cf.get_segment_distribution(res)
|
|
all_querynodes = [node["identifier"] for node in ms.query_nodes]
|
|
all_querynodes = sorted(all_querynodes,
|
|
key=lambda x: len(segment_distribution[x]["sealed"])
|
|
if x in segment_distribution else 0, reverse=True)
|
|
src_node_id = all_querynodes[0]
|
|
dst_node_ids = all_querynodes[1:]
|
|
# add segment ids which are not exist
|
|
sealed_segment_ids = [sealed_segment_id
|
|
for sealed_segment_id in range(max(segment_distribution[src_node_id]["sealed"]) + 100,
|
|
max(segment_distribution[src_node_id]["sealed"]) + 103)]
|
|
# load balance
|
|
self.utility_wrap.load_balance(collection_w.name, src_node_id, dst_node_ids, sealed_segment_ids,
|
|
check_task=CheckTasks.err_res,
|
|
check_items={ct.err_code: 1, ct.err_msg: "not found in source node"})
|
|
|
|
@pytest.mark.tags(CaseLabel.L2)
|
|
def test_load_balance_in_one_group(self):
|
|
"""
|
|
target: test load balance of collection in one group
|
|
method: init a collection, load with multi replicas and load balance among the querynodes in one group
|
|
expected: load balance successfully
|
|
"""
|
|
self._connect()
|
|
querynode_num = len(MilvusSys().query_nodes)
|
|
if querynode_num < 3:
|
|
pytest.skip("skip load balance for multi replicas testcase when querynode number less than 3")
|
|
# init a collection
|
|
c_name = cf.gen_unique_str(prefix)
|
|
collection_w = self.init_collection_wrap(name=c_name)
|
|
nb = 3000
|
|
df = cf.gen_default_dataframe_data(nb)
|
|
collection_w.insert(df)
|
|
# get sealed segments
|
|
collection_w.num_entities
|
|
collection_w.create_index(ct.default_float_vec_field_name, index_params=ct.default_flat_index)
|
|
collection_w.load(replica_number=2)
|
|
# get growing segments
|
|
collection_w.insert(df)
|
|
# get replicas information
|
|
res, _ = collection_w.get_replicas()
|
|
# prepare load balance params
|
|
# find a group which has multi nodes
|
|
group_nodes = []
|
|
for g in res.groups:
|
|
if len(g.group_nodes) >= 2:
|
|
group_nodes = list(g.group_nodes)
|
|
break
|
|
res, _ = self.utility_wrap.get_query_segment_info(c_name)
|
|
segment_distribution = cf.get_segment_distribution(res)
|
|
group_nodes = sorted(group_nodes,
|
|
key=lambda x: len(
|
|
segment_distribution[x]["sealed"])
|
|
if x in segment_distribution else 0, reverse=True)
|
|
src_node_id = group_nodes[0]
|
|
dst_node_ids = group_nodes[1:]
|
|
sealed_segment_ids = segment_distribution[src_node_id]["sealed"]
|
|
# load balance
|
|
self.utility_wrap.load_balance(collection_w.name, src_node_id, dst_node_ids, sealed_segment_ids)
|
|
# get segments distribution after load balance
|
|
res, _ = self.utility_wrap.get_query_segment_info(c_name)
|
|
segment_distribution = cf.get_segment_distribution(res)
|
|
sealed_segment_ids_after_load_balance = segment_distribution[src_node_id]["sealed"]
|
|
# assert src node has no sealed segments
|
|
# assert sealed_segment_ids_after_load_balance == []
|
|
des_sealed_segment_ids = []
|
|
for des_node_id in dst_node_ids:
|
|
des_sealed_segment_ids += segment_distribution[des_node_id]["sealed"]
|
|
# assert sealed_segment_ids is subset of des_sealed_segment_ids
|
|
assert set(sealed_segment_ids).issubset(des_sealed_segment_ids)
|
|
|
|
@pytest.mark.tags(CaseLabel.L3)
|
|
def test_load_balance_not_in_one_group(self):
|
|
"""
|
|
target: test load balance of collection in one group
|
|
method: init a collection, load with multi replicas and load balance among the querynodes in different group
|
|
expected: load balance failed
|
|
"""
|
|
# init a collection
|
|
c_name = cf.gen_unique_str(prefix)
|
|
collection_w = self.init_collection_wrap(name=c_name)
|
|
ms = MilvusSys()
|
|
nb = 3000
|
|
df = cf.gen_default_dataframe_data(nb)
|
|
collection_w.insert(df)
|
|
# get sealed segments
|
|
collection_w.num_entities
|
|
collection_w.load(replica_number=2)
|
|
# get growing segments
|
|
collection_w.insert(df)
|
|
# get replicas information
|
|
res, _ = collection_w.get_replicas()
|
|
# prepare load balance params
|
|
all_querynodes = [node["identifier"] for node in ms.query_nodes]
|
|
# find a group which has multi nodes
|
|
group_nodes = []
|
|
for g in res.groups:
|
|
if len(g.group_nodes) >= 2:
|
|
group_nodes = list(g.group_nodes)
|
|
break
|
|
src_node_id = group_nodes[0]
|
|
dst_node_ids = list(set(all_querynodes) - set(group_nodes))
|
|
res, _ = self.utility_wrap.get_query_segment_info(c_name)
|
|
segment_distribution = cf.get_segment_distribution(res)
|
|
sealed_segment_ids = segment_distribution[src_node_id]["sealed"]
|
|
# load balance
|
|
self.utility_wrap.load_balance(collection_w.name, src_node_id, dst_node_ids, sealed_segment_ids,
|
|
check_task=CheckTasks.err_res,
|
|
check_items={ct.err_code: 1, ct.err_msg: "must be in the same replica group"})
|
|
|
|
@pytest.mark.tags(CaseLabel.L1)
|
|
@pytest.mark.skip(reason="querycoordv2")
|
|
def test_handoff_query_search(self):
|
|
"""
|
|
target: test query search after handoff
|
|
method: 1.load collection
|
|
2.insert, query and search
|
|
3.flush collection and triggere handoff
|
|
4. search with handoff indexed segments
|
|
expected: Search ids before and after handoff are different, because search from growing and search from index
|
|
"""
|
|
collection_w = self.init_collection_wrap(name=cf.gen_unique_str(prefix), shards_num=1)
|
|
collection_w.create_index(default_field_name, default_index_params)
|
|
collection_w.load()
|
|
|
|
# handoff: insert and flush one segment
|
|
df = cf.gen_default_dataframe_data()
|
|
insert_res, _ = collection_w.insert(df)
|
|
term_expr = f'{ct.default_int64_field_name} in {insert_res.primary_keys[:10]}'
|
|
res = df.iloc[:10, :1].to_dict('records')
|
|
collection_w.query(term_expr, check_task=CheckTasks.check_query_results,
|
|
check_items={'exp_res': res,
|
|
"pk_name": collection_w.primary_field.name})
|
|
search_res_before, _ = collection_w.search(df[ct.default_float_vec_field_name][:1].to_list(),
|
|
ct.default_float_vec_field_name,
|
|
ct.default_search_params, ct.default_limit)
|
|
log.debug(collection_w.num_entities)
|
|
|
|
start = time.time()
|
|
while True:
|
|
time.sleep(2)
|
|
segment_infos, _ = self.utility_wrap.get_query_segment_info(collection_w.name)
|
|
# handoff done
|
|
if len(segment_infos) == 1 and segment_infos[0].state == SegmentState.Sealed:
|
|
break
|
|
if time.time() - start > 20:
|
|
raise MilvusException(1, f"Get query segment info after handoff cost more than 20s")
|
|
|
|
# query and search from handoff segments
|
|
collection_w.query(term_expr, check_task=CheckTasks.check_query_results,
|
|
check_items={'exp_res': res,
|
|
"pk_name": collection_w.primary_field.name})
|
|
search_res_after, _ = collection_w.search(df[ct.default_float_vec_field_name][:1].to_list(),
|
|
ct.default_float_vec_field_name,
|
|
ct.default_search_params, ct.default_limit)
|
|
# the ids between twice search is different because of index building
|
|
# log.debug(search_res_before[0].ids)
|
|
# log.debug(search_res_after[0].ids)
|
|
assert search_res_before[0].ids != search_res_after[0].ids
|
|
|
|
# assert search result includes the nq-vector before or after handoff
|
|
assert search_res_after[0].ids[0] == 0
|
|
assert search_res_before[0].ids[0] == search_res_after[0].ids[0]
|
|
|
|
|
|
@pytest.mark.tags(CaseLabel.L3)
|
|
class TestUtilityFlushAll(TestcaseBase):
|
|
|
|
@pytest.mark.parametrize("with_db", [True, False])
|
|
def test_flush_all_multi_collections(self, with_db):
|
|
"""
|
|
target: test flush multi collections
|
|
method: 1.create multi collections
|
|
2. insert data into each collections without flushing
|
|
3. delete some data
|
|
4. flush all collections
|
|
5. verify num entities
|
|
6. create index -> load -> query deleted ids -> query no-deleted ids
|
|
expected: the insert and delete data of all collections are flushed
|
|
"""
|
|
collection_num = 5
|
|
collection_names = []
|
|
delete_num = 100
|
|
|
|
self._connect()
|
|
|
|
if with_db:
|
|
db_name = cf.gen_unique_str("db")
|
|
self.database_wrap.create_database(db_name)
|
|
self.database_wrap.using_database(db_name)
|
|
|
|
for i in range(collection_num):
|
|
collection_w, _, _, insert_ids, _ = self.init_collection_general(prefix, insert_data=True, is_flush=False,
|
|
is_index=True)
|
|
collection_w.delete(f'{ct.default_int64_field_name} in {insert_ids[:delete_num]}')
|
|
collection_names.append(collection_w.name)
|
|
|
|
self.utility_wrap.flush_all(timeout=300)
|
|
cw = ApiCollectionWrapper()
|
|
for c in collection_names:
|
|
cw.init_collection(name=c)
|
|
assert cw.num_entities_without_flush == ct.default_nb
|
|
|
|
cw.create_index(ct.default_float_vec_field_name, ct.default_flat_index)
|
|
cw.load()
|
|
cw.query(f'{ct.default_int64_field_name} in {insert_ids[:100]}', check_task=CheckTasks.check_query_empty)
|
|
|
|
res, _ = cw.query(f'{ct.default_int64_field_name} not in {insert_ids[:delete_num]}')
|
|
assert len(res) == ct.default_nb - delete_num
|
|
|
|
def test_flush_all_db_collections(self):
|
|
"""
|
|
target: test flush all db's collections
|
|
method: 1. create collections and insert data in multi dbs
|
|
2. flush all
|
|
3. verify num entities
|
|
expected: the insert and delete data of all db's collections are flushed
|
|
"""
|
|
tmp_nb = 100
|
|
db_num = 3
|
|
collection_num = 2
|
|
collection_names = {}
|
|
delete_num = 10
|
|
|
|
self._connect()
|
|
for d in range(db_num):
|
|
db_name = cf.gen_unique_str("db")
|
|
self.database_wrap.create_database(db_name)
|
|
self.database_wrap.using_database(db_name)
|
|
db_collection_names = []
|
|
for i in range(collection_num):
|
|
# create collection
|
|
collection_w = self.init_collection_wrap()
|
|
# insert
|
|
insert_res, _ = collection_w.insert(cf.gen_default_dataframe_data(tmp_nb))
|
|
# create index
|
|
collection_w.create_index(ct.default_float_vec_field_name, ct.default_flat_index)
|
|
# delete
|
|
collection_w.delete(f'{ct.default_int64_field_name} in {insert_res.primary_keys[:delete_num]}')
|
|
db_collection_names.append(collection_w.name)
|
|
collection_names[db_name] = db_collection_names
|
|
self.utility_wrap.list_collections()
|
|
|
|
# flush all db collections
|
|
self.utility_wrap.flush_all(timeout=600)
|
|
|
|
for _db, _db_collection_names in collection_names.items():
|
|
self.database_wrap.using_database(_db)
|
|
self.utility_wrap.list_collections()
|
|
for c in _db_collection_names:
|
|
cw = ApiCollectionWrapper()
|
|
cw.init_collection(name=c)
|
|
assert cw.num_entities_without_flush == tmp_nb
|
|
|
|
cw.create_index(ct.default_float_vec_field_name, ct.default_flat_index)
|
|
cw.load()
|
|
cw.query(f'{ct.default_int64_field_name} in {insert_res.primary_keys[:delete_num]}',
|
|
check_task=CheckTasks.check_query_empty)
|
|
|
|
res, _ = cw.query(f'{ct.default_int64_field_name} not in {insert_res.primary_keys[:delete_num]}')
|
|
assert len(res) == tmp_nb - delete_num
|
|
|
|
def test_flush_empty_db_collections(self):
|
|
"""
|
|
target: test flush empty db collections
|
|
method: 1. create collection and insert in the default db
|
|
2. create a db and using the db
|
|
3. flush all
|
|
expected: verify default db's collection flushed
|
|
"""
|
|
tmp_nb = 10
|
|
self._connect()
|
|
|
|
# create collection
|
|
self.database_wrap.using_database(ct.default_db)
|
|
collection_w = self.init_collection_wrap(cf.gen_unique_str(prefix))
|
|
# insert
|
|
insert_res, _ = collection_w.insert(cf.gen_default_dataframe_data(tmp_nb))
|
|
|
|
# create a db and using db
|
|
db_name = cf.gen_unique_str("db")
|
|
self.database_wrap.create_database(db_name)
|
|
self.database_wrap.using_database(db_name)
|
|
|
|
# flush all in the db and no exception
|
|
self.utility_wrap.flush_all(timeout=120)
|
|
|
|
# verify default db's collection not flushed
|
|
self.database_wrap.using_database(ct.default_db)
|
|
assert collection_w.num_entities_without_flush == tmp_nb
|
|
|
|
def test_flush_all_multi_shards_timeout(self):
|
|
"""
|
|
target: test flush_all collection with max shards_num
|
|
method: 1.create collection with max shards_num
|
|
2.insert data
|
|
3.flush all with a small timeout and gets exception
|
|
4.flush and verify num entities
|
|
expected:
|
|
"""
|
|
collection_w = self.init_collection_wrap(name=cf.gen_unique_str(prefix), shards_num=ct.max_shards_num)
|
|
df = cf.gen_default_dataframe_data()
|
|
collection_w.insert(df)
|
|
error = {ct.err_code: 1, ct.err_msg: "wait for flush all timeout"}
|
|
self.utility_wrap.flush_all(timeout=5, check_task=CheckTasks.err_res, check_items=error)
|
|
self.utility_wrap.flush_all(timeout=120)
|
|
assert collection_w.num_entities_without_flush == ct.default_nb
|
|
|
|
def test_flush_all_no_collections(self):
|
|
"""
|
|
target: test flush all without any collections
|
|
method: connect and flush all
|
|
expected: no exception
|
|
"""
|
|
self._connect()
|
|
self.utility_wrap.flush_all(check_task=None)
|
|
|
|
def test_flush_all_async(self):
|
|
"""
|
|
target: test flush all collections with _async
|
|
method: flush all collections and _async=True
|
|
expected: finish flush all collection within a period of time
|
|
"""
|
|
collection_num = 5
|
|
collection_names = []
|
|
delete_num = 100
|
|
|
|
for i in range(collection_num):
|
|
collection_w, _, _, insert_ids, _ = self.init_collection_general(prefix, insert_data=True, is_flush=False,
|
|
is_index=True)
|
|
collection_w.delete(f'{ct.default_int64_field_name} in {insert_ids[:delete_num]}')
|
|
collection_names.append(collection_w.name)
|
|
|
|
self.utility_wrap.flush_all(_async=True)
|
|
_async_timeout = 60
|
|
cw = ApiCollectionWrapper()
|
|
start = time.time()
|
|
while time.time() - start < _async_timeout:
|
|
time.sleep(2.0)
|
|
flush_flag = False
|
|
for c in collection_names:
|
|
cw.init_collection(name=c)
|
|
if cw.num_entities_without_flush == ct.default_nb:
|
|
flush_flag = True
|
|
else:
|
|
log.debug(f"collection num entities: {cw.num_entities_without_flush} of collection {c}")
|
|
flush_flag = False
|
|
if flush_flag:
|
|
break
|
|
if time.time() - start > _async_timeout:
|
|
raise MilvusException(1, f"Waiting more than {_async_timeout}s for the flush all")
|
|
|
|
def test_flush_all_while_insert_delete(self):
|
|
"""
|
|
target: test flush all while insert and delete
|
|
method: 1. prepare multi collections
|
|
2. flush_all while inserting and deleting
|
|
expected:
|
|
"""
|
|
from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED
|
|
collection_num = 5
|
|
collection_names = []
|
|
delete_num = 100
|
|
delete_ids = [_i for _i in range(delete_num)]
|
|
|
|
for i in range(collection_num):
|
|
collection_w = self.init_collection_wrap(name=cf.gen_unique_str(prefix), shards_num=4)
|
|
df = cf.gen_default_dataframe_data(nb=ct.default_nb, start=0)
|
|
collection_w.insert(df)
|
|
collection_names.append(collection_w.name)
|
|
|
|
def do_insert():
|
|
cw = ApiCollectionWrapper()
|
|
df = cf.gen_default_dataframe_data(nb=ct.default_nb, start=ct.default_nb)
|
|
for c_name in collection_names:
|
|
cw.init_collection(c_name)
|
|
insert_res, _ = cw.insert(df)
|
|
assert insert_res.insert_count == ct.default_nb
|
|
|
|
def do_delete():
|
|
cw = ApiCollectionWrapper()
|
|
for c_name in collection_names:
|
|
cw.init_collection(c_name)
|
|
del_res, _ = cw.delete(f"{ct.default_int64_field_name} in {delete_ids}")
|
|
assert del_res.delete_count == delete_num
|
|
|
|
def do_flush_all():
|
|
time.sleep(2)
|
|
self.utility_wrap.flush_all(timeout=600)
|
|
|
|
executor = ThreadPoolExecutor(max_workers=3)
|
|
insert_task = executor.submit(do_insert)
|
|
delete_task = executor.submit(do_delete)
|
|
flush_task = executor.submit(do_flush_all)
|
|
|
|
# wait all tasks completed
|
|
wait([insert_task, delete_task, flush_task], return_when=ALL_COMPLETED)
|
|
|
|
# verify
|
|
for c in collection_names:
|
|
cw = ApiCollectionWrapper()
|
|
cw.init_collection(name=c)
|
|
log.debug(f"num entities: {cw.num_entities_without_flush}")
|
|
assert cw.num_entities_without_flush >= ct.default_nb
|
|
assert cw.num_entities_without_flush <= ct.default_nb * 2
|
|
|
|
cw.create_index(ct.default_float_vec_field_name, ct.default_flat_index)
|
|
cw.load()
|
|
cw.query(f'{ct.default_int64_field_name} in {delete_ids}', check_task=CheckTasks.check_query_empty)
|
|
|
|
res, _ = cw.query(f'{ct.default_int64_field_name} not in {delete_ids}')
|
|
assert len(res) == ct.default_nb * 2 - delete_num
|
|
|