test: add Go, PyMilvus, and REST e2e for FMINDEX and membership_match (#52629)

## Summary

Add end-to-end coverage for FMINDEX and the unified `membership_match`
expression across the Go SDK, PyMilvus, and RESTful v2.

Related issues: #51375, #52778, #53036

## Coverage

### FMINDEX

- Build parameter defaults, boundaries, combined parameters, and invalid
types or ranges.
- VARCHAR-only validation, including JSON, scalar, and struct-array
sub-fields.
- Prefix, infix, suffix, general LIKE recheck, regex fallback, escaped
wildcards, NULL, and UTF-8 behavior.
- Long-text low-hit fixtures and sealed segments above
`minSegmentNumRowsToEnableIndex`, preventing fake-finished index
coverage.
- Exact indexed-field versus unindexed twin-field PK comparison.
- Growing plus sealed visibility, mutation, upsert, named-index drop,
and rebuild lifecycle checks.
- Query and Search coverage with exact projected fields, distance
ordering, index identity, type, and row-progress assertions.
- VARCHAR AUTOINDEX resolution is verified against the configured
scalar-index policy without hard-coding a specific index type.

### membership_match

- Bloom and Roaring wire compatibility through a shared static manifest
while each SDK keeps independent collection setup.
- Bloom zero-false-negative checks plus a statistically bounded
false-positive check tied to the configured FPR; Roaring uses exact PK
ground truth.
- Query and Search must return the same complete Bloom-filtered PK set
when topK covers the fixture.
- Integer widths, negative values, INT64 boundaries, VARCHAR, JSON
strict typing, nested paths, dynamic fields, NULL semantics, and
unsupported top-level BOOL/FLOAT fields.
- Positive, negated, empty, malformed, domain-mismatch, delete, growing
plus sealed, and struct-array rejection cases.
- Explicit `STL_SORT`, `INVERTED`, `BITMAP`, and `TRIE` scalar-index
paths.
- Internal-only `HYBRID` paths are exercised through the supported
`AUTOINDEX` request and verified against DataCoord's resolved build
parameters for primitive and JSON-path indexes.
- Mutation tests use isolated collections; shared read-only fixtures
register cleanup before setup begins.

### RESTful v2

- FMINDEX anchored LIKE Query and Search compared with exact PK ground
truth and an unindexed twin field.
- Index identity, type, persisted parameters, row progress, completion,
and collection load state are verified before assertions.
- Search validates exact projected values by primary key.
- JSON string and missing membership template parameters are verified as
server-side validation failures.
- Python `bytes` are verified to fail local JSON serialization before
HTTP. A positive REST binary-template case remains blocked by #53036.

## Validation

- `go test -tags dynamic,test -gcflags='all=-N -l' -c ./testcases`
- `go vet -tags dynamic,test ./testcases`
- `go test ./membership/sbbf -count=1`
- Ruff check and format check for all changed Python and REST files
- Python byte-compilation for all changed Python and REST files
- Pytest collection: 68 PyMilvus/FMINDEX tests and 7 REST tests
- Cross-SDK membership wire-manifest verification
- `git diff --check`

The Proxy parameter-resolution unit test and Bloom integration package
cannot execute on the host because it lacks the native `milvus_core`,
`rdkafka`, and `rocksdb` pkg-config packages. A targeted PyMilvus
FMINDEX run reached index creation but timed out after 120 seconds
before assertions; its temporary collection was cleaned up. Full
server-backed E2E was not completed.

Signed-off-by: Yanliang Qiao <yanliang.qiao@zilliz.com>
This commit is contained in:
yanliang567 authored and GitHub committed 2026-09-07 14:07:30 +08:00
1 parent c8e219ddae
commit 3067e2542d
17 files changed
+5291 -119

No files matched your search

+18
View File
@@ -1659,6 +1659,24 @@ func Test_ngram_parseIndexParams(t *testing.T) {
}
func Test_fmindex_parseIndexParams(t *testing.T) {
t.Run("varchar autoindex follows configured scalar policy", func(t *testing.T) {
cit := &createIndexTask{
req: &milvuspb.CreateIndexRequest{
ExtraParams: []*commonpb.KeyValuePair{
{Key: common.IndexTypeKey, Value: AutoIndexName},
},
},
fieldSchema: &schemapb.FieldSchema{
FieldID: 101, Name: "FieldID", DataType: schemapb.DataType_VarChar,
},
}
err := cit.parseIndexParams(context.TODO())
assert.NoError(t, err)
resolvedType, err := funcutil.GetAttrByKeyFromRepeatedKV(common.IndexTypeKey, cit.newIndexParams)
assert.NoError(t, err)
assert.Equal(t, Params.AutoIndexConfig.ScalarVarcharIndexType.GetValue(), resolvedType)
})
t.Run("valid fmindex index params without sample rate", func(t *testing.T) {
cit := &createIndexTask{
req: &milvuspb.CreateIndexRequest{
+16
View File
@@ -0,0 +1,16 @@
{
"schema_version": 1,
"bloom": {
"domain": "int64",
"members": [0, 1, 2, 42],
"present_members": [0, 1, 2, 42],
"fpr": 0.001,
"hex": "4d424631010001000400000000000000fca9f1d24d62503f01000000010000000003100880000406002400420040060210044004008000984040002830410000"
},
"roaring": {
"domain": "int64",
"members": [-2, -1, 0, 1, 42],
"present_members": [0, 1, 42],
"hex": "4d5242310100010005000000000000003a0000000000000000000000000000000200000000000000000000003a300000010000000000020010000000000001002a00ffffffff3a30000001000000ffff010010000000feffffff"
}
}
+31
View File
@@ -0,0 +1,31 @@
// Licensed to the LF AI & Data foundation under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package testcases
import (
"context"
"github.com/milvus-io/milvus/tests/go_client/base"
)
// Type aliases shared across all test files to keep test signatures readable.
// They live here (not in a feature-specific file) so no file depends on another
// test file for a cross-cutting declaration.
type (
CtxT = context.Context
MC = *base.MilvusClient
)
@@ -0,0 +1,798 @@
// Licensed to the LF AI & Data foundation under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// Bloom-kind membership_match coverage through the public Go SDK.
package testcases
import (
"context"
"fmt"
"math"
"testing"
"time"
"github.com/stretchr/testify/require"
"github.com/milvus-io/milvus/client/v3/column"
"github.com/milvus-io/milvus/client/v3/entity"
"github.com/milvus-io/milvus/client/v3/index"
client "github.com/milvus-io/milvus/client/v3/milvusclient"
"github.com/milvus-io/milvus/tests/go_client/common"
hp "github.com/milvus-io/milvus/tests/go_client/testcases/helper"
)
func queryBloomIDs(t *testing.T, ctx CtxT, mc MC, collectionName, expr string, blob any) []int64 {
t.Helper()
return queryMembershipIDs(t, ctx, mc, collectionName, expr, "bf", blob)
}
func queryBloomIDSet(t *testing.T, ctx CtxT, mc MC, collectionName, expr string, blob any) map[int64]struct{} {
t.Helper()
return queryMembershipIDSet(t, ctx, mc, collectionName, expr, "bf", blob)
}
func requireBloomFPRBound(
t *testing.T,
got map[int64]struct{},
expected map[int64]struct{},
totalRows int,
configuredFPR float64,
) {
t.Helper()
require.Greater(t, configuredFPR, 0.0)
require.Less(t, configuredFPR, 1.0)
require.LessOrEqual(t, len(expected), totalRows)
for pk := range expected {
require.Containsf(t, got, pk, "Bloom result dropped true member id=%d", pk)
}
nonMembers := totalRows - len(expected)
falsePositives := len(got) - len(expected)
mean := float64(nonMembers) * configuredFPR
standardDeviation := math.Sqrt(float64(nonMembers) * configuredFPR * (1 - configuredFPR))
upperBound := int(math.Ceil(mean + 6*standardDeviation))
require.LessOrEqualf(t, falsePositives, upperBound,
"Bloom result has %d false positives over %d unique non-members; configured fpr=%.4f, six-sigma upper bound=%d",
falsePositives, nonMembers, configuredFPR, upperBound)
}
// TestBloomMatchZeroFalseNegatives verifies bloom_match has no row-level false
// negatives: exact `in` result is a subset of bloom_match result, and
// bloom_match plus its negation partition every non-NULL row. Query and Search
// intentionally reuse this same loaded collection and filter blob.
func TestBloomMatchZeroFalseNegatives(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("bloom_zero_fn", 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRows(t, ctx, mc, collectionName, false)
flushLoadMembership(t, ctx, mc, collectionName)
members := []int64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
blob, err := client.NewBloomFilterBlob(members, 0.001)
require.NoError(t, err)
exact := queryBloomIDSet(t, ctx, mc, collectionName,
fmt.Sprintf("%s in [0,1,2,3,4,5,6,7,8,9]", membershipCreatorField), nil)
got := queryBloomIDSet(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {bf}, type=bloom)", membershipCreatorField), blob)
require.NotEmpty(t, exact)
requireBloomResult(t, got, exact, membershipTotalRows, "sealed INT64 Query")
searchGot := searchMembershipIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {bf}, type=bloom)", membershipCreatorField), "bf", blob)
searchSet := make(map[int64]struct{}, len(searchGot))
for _, pk := range searchGot {
searchSet[pk] = struct{}{}
}
require.Len(t, searchSet, len(searchGot), "Search membership_match returned duplicate PKs")
require.Equal(t, got, searchSet, "Search membership_match must return the same Bloom-filtered PK set as Query")
notRs := queryBloomIDs(t, ctx, mc, collectionName,
fmt.Sprintf("not membership_match(%s, {bf}, type=bloom)", membershipCreatorField), blob)
notSet := int64IDSet(notRs)
require.Len(t, notSet, len(notRs), "not bloom_match returned duplicate PKs")
for id := int64(0); id < membershipTotalRows; id++ {
_, matched := got[id]
_, negated := notSet[id]
require.NotEqualf(t, matched, negated,
"bloom_match and its negation must partition every row; id=%d matched=%t negated=%t",
id, matched, negated)
}
}
// TestBloomMatchVarcharAndDomainMismatch verifies varchar membership has no
// false negatives and that a blob built from the wrong value domain is rejected.
func TestBloomMatchVarcharAndDomainMismatch(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("bloom_varchar", 6)
schema := entity.NewSchema().WithName(collectionName).
WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
WithField(entity.NewField().WithName("tag").WithDataType(entity.FieldTypeVarChar).WithMaxLength(64)).
WithField(entity.NewField().WithName(membershipCreatorField).WithDataType(entity.FieldTypeInt64)).
WithField(entity.NewField().WithName(membershipVectorField).WithDataType(entity.FieldTypeFloatVector).WithDim(membershipVectorDim))
require.NoError(t, mc.CreateCollection(ctx, client.NewCreateCollectionOption(collectionName, schema).
WithConsistencyLevel(entity.ClStrong)))
t.Cleanup(func() {
cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
require.NoError(t, mc.DropCollection(cleanupCtx, client.NewDropCollectionOption(collectionName)))
})
ids := make([]int64, membershipTotalRows)
tags := make([]string, membershipTotalRows)
creators := make([]int64, membershipTotalRows)
vectors := make([][]float32, membershipTotalRows)
for i := 0; i < membershipTotalRows; i++ {
ids[i] = int64(i)
tags[i] = fmt.Sprintf("tag%d", i%membershipDomain)
creators[i] = int64(i % membershipDomain)
v := make([]float32, membershipVectorDim)
v[0] = float32(i)
vectors[i] = v
}
_, err := mc.Insert(ctx, client.NewColumnBasedInsertOption(collectionName).
WithInt64Column("id", ids).
WithVarcharColumn("tag", tags).
WithInt64Column(membershipCreatorField, creators).
WithFloatVectorColumn(membershipVectorField, membershipVectorDim, vectors))
require.NoError(t, err)
flushLoadMembership(t, ctx, mc, collectionName)
strMembers := []string{"tag0", "tag1", "tag2", "tag3", "tag4"}
strBlob, err := client.NewBloomFilterBlob(strMembers, 0.001)
require.NoError(t, err)
exact := queryBloomIDSet(t, ctx, mc, collectionName,
`tag in ["tag0","tag1","tag2","tag3","tag4"]`, nil)
bloomSet := queryBloomIDSet(t, ctx, mc, collectionName, "membership_match(tag, {bf}, type=bloom)", strBlob)
requireBloomResult(t, bloomSet, exact, membershipTotalRows, "VARCHAR Query")
// int64-domain blob on a varchar field must be rejected
intBlob, err := client.NewBloomFilterBlob([]int64{0, 1, 2}, 0.001)
require.NoError(t, err)
_, err = mc.Query(ctx, client.NewQueryOption(collectionName).
WithFilter("membership_match(tag, {bf}, type=bloom)").WithOutputFields("id").
WithTemplateParam("bf", intBlob).
WithConsistencyLevel(entity.ClStrong))
common.CheckErr(t, err, false, "value domain but field")
// utf8-domain blob on an int64 field must be rejected
_, err = mc.Query(ctx, client.NewQueryOption(collectionName).
WithFilter(fmt.Sprintf("membership_match(%s, {bf}, type=bloom)", membershipCreatorField)).WithOutputFields("id").
WithTemplateParam("bf", strBlob).
WithConsistencyLevel(entity.ClStrong))
common.CheckErr(t, err, false, "value domain but field")
}
// TestBloomMatchMalformedBlobRejected verifies a malformed blob and a literal
// array argument are rejected rather than silently unfiltered.
func TestBloomMatchMalformedBlobRejected(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("bloom_malformed", 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRows(t, ctx, mc, collectionName, false)
flushLoadMembership(t, ctx, mc, collectionName)
// malformed MBF1 blob
_, err := mc.Query(ctx, client.NewQueryOption(collectionName).
WithFilter(fmt.Sprintf("membership_match(%s, {bf}, type=bloom)", membershipCreatorField)).WithOutputFields("id").
WithTemplateParam("bf", client.BloomFilterBlob([]byte("not-a-real-blob"))).
WithConsistencyLevel(entity.ClStrong))
common.CheckErr(t, err, false, "unknown format magic")
// literal array argument is not a pre-built blob
_, err = mc.Query(ctx, client.NewQueryOption(collectionName).
WithFilter(fmt.Sprintf("membership_match(%s, [1,2,3], type=bloom)", membershipCreatorField)).WithOutputFields("id").
WithConsistencyLevel(entity.ClStrong))
common.CheckErr(t, err, false, "must be a {template} placeholder")
}
// TestBloomMatchNullRowsFoldToFalse verifies NULL rows never match bloom_match
// nor its negation (NULL folds to FALSE on both sides).
func TestBloomMatchNullRowsFoldToFalse(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("bloom_null", 6)
schema := entity.NewSchema().WithName(collectionName).
WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
WithField(entity.NewField().WithName(membershipCreatorField).WithDataType(entity.FieldTypeInt64).WithNullable(true)).
WithField(entity.NewField().WithName(membershipVectorField).WithDataType(entity.FieldTypeFloatVector).WithDim(membershipVectorDim))
require.NoError(t, mc.CreateCollection(ctx, client.NewCreateCollectionOption(collectionName, schema).
WithConsistencyLevel(entity.ClStrong)))
t.Cleanup(func() {
cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
require.NoError(t, mc.DropCollection(cleanupCtx, client.NewDropCollectionOption(collectionName)))
})
ids := make([]int64, membershipTotalRows)
values := make([]int64, 0, membershipTotalRows)
valid := make([]bool, membershipTotalRows)
vectors := make([][]float32, membershipTotalRows)
for i := 0; i < membershipTotalRows; i++ {
ids[i] = int64(i)
v := make([]float32, membershipVectorDim)
v[0] = float32(i)
vectors[i] = v
if i%8 == 7 {
valid[i] = false
continue
}
valid[i] = true
values = append(values, int64(i%membershipDomain))
}
col, err := column.NewNullableColumnInt64(membershipCreatorField, values, valid)
require.NoError(t, err)
_, err = mc.Insert(ctx, client.NewColumnBasedInsertOption(collectionName).
WithInt64Column("id", ids).
WithColumns(col).
WithFloatVectorColumn(membershipVectorField, membershipVectorDim, vectors))
require.NoError(t, err)
flushLoadMembership(t, ctx, mc, collectionName)
members := make([]int64, 0, membershipDomain)
for v := 0; v < membershipDomain; v++ {
members = append(members, int64(v))
}
blob, err := client.NewBloomFilterBlob(members, 0.001)
require.NoError(t, err)
got := queryBloomIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {bf}, type=bloom)", membershipCreatorField), blob)
expected := make([]int64, 0, membershipTotalRows-membershipTotalRows/8)
for id := 0; id < membershipTotalRows; id++ {
if id%8 != 7 {
expected = append(expected, int64(id))
}
}
require.ElementsMatch(t, expected, got, "all and only non-NULL rows must match when every value is a member")
notRs := queryBloomIDs(t, ctx, mc, collectionName,
fmt.Sprintf("not membership_match(%s, {bf}, type=bloom)", membershipCreatorField), blob)
require.Empty(t, notRs, "NULL folds to false and every non-NULL value matches, so negation must match nothing")
}
// TestBloomMatchScalarIndexTypeMatrix verifies bloom_match stays correct when a
// scalar index is built on the field before load. STL_SORT reports
// has_raw_data=true, so the loader drops the raw column and bloom_match must
// recover each value via the index's per-row Reverse_Lookup; INVERTED and
// BITMAP cover the other supported scalar-index entry points. All must remain
// zero-false-negative.
func TestBloomMatchScalarIndexTypeMatrix(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
members := []int64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
blob, err := client.NewBloomFilterBlob(members, 0.001)
require.NoError(t, err)
for _, indexType := range []string{"STL_SORT", "INVERTED", "BITMAP"} {
t.Run(indexType, func(t *testing.T) {
collectionName := common.GenRandomString("bloom_idx_"+indexType, 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRows(t, ctx, mc, collectionName, false)
flushTask, err := mc.Flush(ctx, client.NewFlushOption(collectionName))
require.NoError(t, err)
require.NoError(t, flushTask.Await(ctx))
// Build the scalar index on creator_id BEFORE the first load so the
// loader materializes it and (for STL_SORT) drops the raw column.
var scalarIndex index.Index
switch indexType {
case "STL_SORT":
scalarIndex = index.NewSortedIndex()
case "INVERTED":
scalarIndex = index.NewInvertedIndex()
case "BITMAP":
scalarIndex = index.NewBitmapIndex()
}
idxTask, err := mc.CreateIndex(ctx, client.NewCreateIndexOption(collectionName, membershipCreatorField,
scalarIndex))
require.NoError(t, err)
require.NoError(t, idxTask.Await(ctx))
vecTask, err := mc.CreateIndex(ctx, client.NewCreateIndexOption(collectionName, membershipVectorField,
index.NewFlatIndex(entity.L2)))
require.NoError(t, err)
require.NoError(t, vecTask.Await(ctx))
loadTask, err := mc.LoadCollection(ctx, client.NewLoadCollectionOption(collectionName))
require.NoError(t, err)
require.NoError(t, loadTask.Await(ctx))
exact := queryBloomIDSet(t, ctx, mc, collectionName,
fmt.Sprintf("%s in [0,1,2,3,4,5,6,7,8,9]", membershipCreatorField), nil)
got := queryBloomIDSet(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {bf}, type=bloom)", membershipCreatorField), blob)
require.NotEmpty(t, exact)
requireBloomResult(t, got, exact, membershipTotalRows, "explicit scalar index "+indexType)
})
}
}
// TestBloomMatchAutoIndex verifies bloom_match stays correct when the field is
// indexed with AUTOINDEX (the default entry point when no index type is
// specified). Low- and high-cardinality datasets are both covered without
// coupling the assertion to AUTOINDEX's internal implementation choice.
func TestBloomMatchAutoIndex(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
members := []int64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
blob, err := client.NewBloomFilterBlob(members, 0.001)
require.NoError(t, err)
// Exercise AUTOINDEX with both high- and low-cardinality inputs. Explicit
// index-path coverage lives in TestBloomMatchScalarIndexTypeMatrix.
for _, domain := range []int{200, 50} {
t.Run(fmt.Sprintf("domain_%d", domain), func(t *testing.T) {
collectionName := common.GenRandomString("bloom_auto", 6)
schema := entity.NewSchema().WithName(collectionName).
WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
WithField(entity.NewField().WithName(membershipCreatorField).WithDataType(entity.FieldTypeInt64)).
WithField(entity.NewField().WithName(membershipVectorField).WithDataType(entity.FieldTypeFloatVector).WithDim(membershipVectorDim))
require.NoError(t, mc.CreateCollection(ctx, client.NewCreateCollectionOption(collectionName, schema).
WithConsistencyLevel(entity.ClStrong)))
t.Cleanup(func() {
cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
require.NoError(t, mc.DropCollection(cleanupCtx, client.NewDropCollectionOption(collectionName)))
})
ids := make([]int64, membershipTotalRows)
creators := make([]int64, membershipTotalRows)
vectors := make([][]float32, membershipTotalRows)
for i := 0; i < membershipTotalRows; i++ {
ids[i] = int64(i)
creators[i] = int64(i % domain)
v := make([]float32, membershipVectorDim)
v[0] = float32(i)
vectors[i] = v
}
_, err := mc.Insert(ctx, client.NewColumnBasedInsertOption(collectionName).
WithInt64Column("id", ids).
WithInt64Column(membershipCreatorField, creators).
WithFloatVectorColumn(membershipVectorField, membershipVectorDim, vectors))
require.NoError(t, err)
flushTask, err := mc.Flush(ctx, client.NewFlushOption(collectionName))
require.NoError(t, err)
require.NoError(t, flushTask.Await(ctx))
idxTask, err := mc.CreateIndex(ctx, client.NewCreateIndexOption(collectionName, membershipCreatorField,
index.NewAutoIndex(entity.L2)))
require.NoError(t, err)
require.NoError(t, idxTask.Await(ctx))
vecTask, err := mc.CreateIndex(ctx, client.NewCreateIndexOption(collectionName, membershipVectorField,
index.NewFlatIndex(entity.L2)))
require.NoError(t, err)
require.NoError(t, vecTask.Await(ctx))
loadTask, err := mc.LoadCollection(ctx, client.NewLoadCollectionOption(collectionName))
require.NoError(t, err)
require.NoError(t, loadTask.Await(ctx))
exact := queryBloomIDSet(t, ctx, mc, collectionName,
fmt.Sprintf("%s in [0,1,2,3,4,5,6,7,8,9]", membershipCreatorField), nil)
got := queryBloomIDSet(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {bf}, type=bloom)", membershipCreatorField), blob)
require.NotEmpty(t, exact)
requireBloomResult(t, got, exact, membershipTotalRows,
fmt.Sprintf("AUTOINDEX domain %d", domain))
})
}
}
// TestBloomMatchRejectedInDelete verifies bloom_match is rejected in a delete
// expression: it has false positives, so a delete driven by it would remove rows
// outside the caller's set (roaring_match, being exact, is allowed instead).
func TestBloomMatchRejectedInDelete(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("bloom_delete", 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRows(t, ctx, mc, collectionName, false)
flushLoadMembership(t, ctx, mc, collectionName)
blob, err := client.NewBloomFilterBlob([]int64{0, 1, 2}, 0.001)
require.NoError(t, err)
_, err = mc.Delete(ctx, client.NewDeleteOption(collectionName).
WithExpr(fmt.Sprintf("membership_match(%s, {bf}, type=bloom)", membershipCreatorField)).
WithTemplateParam("bf", blob))
common.CheckErr(t, err, false, "membership_match with an approximate bloom filter blob cannot be used in delete expressions")
}
// TestBloomMatchIntTypeMatrix verifies one int64-built blob probes every integer
// field width (INT8/16/32/64 all widen to int64 before hashing).
func TestBloomMatchIntTypeMatrix(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("bloom_intmatrix", 6)
schema := entity.NewSchema().WithName(collectionName).
WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
WithField(entity.NewField().WithName("i8").WithDataType(entity.FieldTypeInt8)).
WithField(entity.NewField().WithName("i16").WithDataType(entity.FieldTypeInt16)).
WithField(entity.NewField().WithName("i32").WithDataType(entity.FieldTypeInt32)).
WithField(entity.NewField().WithName("i64").WithDataType(entity.FieldTypeInt64)).
WithField(entity.NewField().WithName(membershipVectorField).WithDataType(entity.FieldTypeFloatVector).WithDim(membershipVectorDim))
require.NoError(t, mc.CreateCollection(ctx, client.NewCreateCollectionOption(collectionName, schema).
WithConsistencyLevel(entity.ClStrong)))
t.Cleanup(func() {
cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
require.NoError(t, mc.DropCollection(cleanupCtx, client.NewDropCollectionOption(collectionName)))
})
ids := make([]int64, membershipTotalRows)
i8 := make([]int8, membershipTotalRows)
i16 := make([]int16, membershipTotalRows)
i32 := make([]int32, membershipTotalRows)
i64 := make([]int64, membershipTotalRows)
vectors := make([][]float32, membershipTotalRows)
for i := 0; i < membershipTotalRows; i++ {
ids[i] = int64(i)
i8[i] = int8(i % membershipDomain)
i16[i] = int16(i % membershipDomain)
i32[i] = int32(i % membershipDomain)
i64[i] = int64(i % membershipDomain)
v := make([]float32, membershipVectorDim)
v[0] = float32(i)
vectors[i] = v
}
_, err := mc.Insert(ctx, client.NewColumnBasedInsertOption(collectionName).
WithInt64Column("id", ids).
WithInt8Column("i8", i8).
WithInt16Column("i16", i16).
WithInt32Column("i32", i32).
WithInt64Column("i64", i64).
WithFloatVectorColumn(membershipVectorField, membershipVectorDim, vectors))
require.NoError(t, err)
flushLoadMembership(t, ctx, mc, collectionName)
blob, err := client.NewBloomFilterBlob([]int64{0, 1, 2, 3, 4}, 0.001)
require.NoError(t, err)
for _, field := range []string{"i8", "i16", "i32", "i64"} {
exact := queryBloomIDSet(t, ctx, mc, collectionName,
fmt.Sprintf("%s in [0,1,2,3,4]", field), nil)
got := queryBloomIDSet(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {bf}, type=bloom)", field), blob)
require.NotEmpty(t, exact)
requireBloomResult(t, got, exact, membershipTotalRows, "integer field "+field)
}
}
// TestBloomMatchJsonPathStrictTyping verifies JSON-path membership is strictly
// typed: int-encoded members match, float-encoded values (e.g. 5.0) never match
// an int64 member 5, and rows missing the key match under neither polarity.
func TestBloomMatchJsonPathStrictTyping(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("bloom_json", 6)
schema := entity.NewSchema().WithName(collectionName).
WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
WithField(entity.NewField().WithName("meta").WithDataType(entity.FieldTypeJSON)).
WithField(entity.NewField().WithName(membershipVectorField).WithDataType(entity.FieldTypeFloatVector).WithDim(membershipVectorDim))
require.NoError(t, mc.CreateCollection(ctx, client.NewCreateCollectionOption(collectionName, schema).
WithConsistencyLevel(entity.ClStrong)))
t.Cleanup(func() {
cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
require.NoError(t, mc.DropCollection(cleanupCtx, client.NewDropCollectionOption(collectionName)))
})
const totalRows = 1000
ids := make([]int64, totalRows)
jsonValues := make([][]byte, totalRows)
vectors := make([][]float32, totalRows)
for i := 0; i < totalRows; i++ {
ids[i] = int64(i)
v := make([]float32, membershipVectorDim)
v[0] = float32(i)
vectors[i] = v
switch {
case i%11 == 0:
jsonValues[i] = []byte(`{"other": 1}`) // missing key
case i%3 == 0:
jsonValues[i] = []byte(fmt.Sprintf(`{"uid": %d.0}`, i%10)) // float-encoded
default:
jsonValues[i] = []byte(fmt.Sprintf(`{"uid": %d}`, i%10)) // int-encoded
}
}
_, err := mc.Insert(ctx, client.NewColumnBasedInsertOption(collectionName).
WithInt64Column("id", ids).
WithColumns(column.NewColumnJSONBytes("meta", jsonValues)).
WithFloatVectorColumn(membershipVectorField, membershipVectorDim, vectors))
require.NoError(t, err)
flushLoadMembership(t, ctx, mc, collectionName)
blob, err := client.NewBloomFilterBlob([]int64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}, 0.001)
require.NoError(t, err)
bloomRes := queryBloomIDs(t, ctx, mc, collectionName, `membership_match(meta["uid"], {bf}, type=bloom)`, blob)
expectedBloom := make([]int64, 0, totalRows)
for id := 0; id < totalRows; id++ {
if id%11 != 0 && id%3 != 0 {
expectedBloom = append(expectedBloom, int64(id))
}
}
require.ElementsMatch(t, expectedBloom, bloomRes,
"all and only int-encoded member rows must match the JSON path")
negatedRes := queryBloomIDs(t, ctx, mc, collectionName, `not membership_match(meta["uid"], {bf}, type=bloom)`, blob)
expectedNegated := make([]int64, 0, totalRows)
for id := 0; id < totalRows; id++ {
if id%11 != 0 && id%3 == 0 {
expectedNegated = append(expectedNegated, int64(id))
}
}
require.ElementsMatch(t, expectedNegated, negatedRes,
"negation must select float-encoded rows but keep missing-key rows invalid")
// The divergence from exact `in`: `in` unifies 5.0 == 5, so it returns the
// float-encoded rows too; bloom_match, strictly typed, does not.
exactRes := queryBloomIDs(t, ctx, mc, collectionName, `meta["uid"] in [0,1,2,3,4,5,6,7,8,9]`, nil)
exactSet := make(map[int64]struct{}, len(exactRes))
for _, id := range exactRes {
exactSet[id] = struct{}{}
}
bloomSet := make(map[int64]struct{}, len(bloomRes))
for _, id := range bloomRes {
bloomSet[id] = struct{}{}
}
floatEncodedPresent := 0
for id := range exactSet {
if id%3 == 0 && id%11 != 0 {
floatEncodedPresent++
_, inBloom := bloomSet[id]
require.Falsef(t, inBloom, "float-encoded row id=%d must be in exact `in` but NOT in bloom_match", id)
}
}
require.Positive(t, floatEncodedPresent, "fixture must contain float-encoded member rows")
}
// TestBloomMatchJsonWholeDocNestedAndDynamicPath verifies the JSON-path forms
// beyond a single key: the whole JSON document, a nested path, and a dynamic
// field (an unknown identifier resolving to a $meta path). Each is strictly
// typed, so a float-encoded value never matches an int64 member.
func TestBloomMatchJsonWholeDocNestedAndDynamicPath(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("bloom_json_forms", 6)
schema := entity.NewSchema().WithName(collectionName).
WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
WithField(entity.NewField().WithName("meta").WithDataType(entity.FieldTypeJSON)).
WithField(entity.NewField().WithName(membershipVectorField).WithDataType(entity.FieldTypeFloatVector).WithDim(membershipVectorDim))
require.NoError(t, mc.CreateCollection(ctx, client.NewCreateCollectionOption(collectionName, schema).
WithConsistencyLevel(entity.ClStrong)))
t.Cleanup(func() {
cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
require.NoError(t, mc.DropCollection(cleanupCtx, client.NewDropCollectionOption(collectionName)))
})
const totalRows = 1000
ids := make([]int64, totalRows)
jsonValues := make([][]byte, totalRows)
vectors := make([][]float32, totalRows)
for i := 0; i < totalRows; i++ {
ids[i] = int64(i)
v := make([]float32, membershipVectorDim)
v[0] = float32(i)
vectors[i] = v
switch i % 3 {
case 0:
// nested object: {"a": {"b": <int>}} — int-encoded
jsonValues[i] = []byte(fmt.Sprintf(`{"a": {"b": %d}}`, i%10))
case 1:
// nested object with float-encoded value
jsonValues[i] = []byte(fmt.Sprintf(`{"a": {"b": %d.0}}`, i%10))
default:
// bare scalar whole-document value (int)
jsonValues[i] = []byte(fmt.Sprintf(`%d`, i%10))
}
}
_, err := mc.Insert(ctx, client.NewColumnBasedInsertOption(collectionName).
WithInt64Column("id", ids).
WithColumns(column.NewColumnJSONBytes("meta", jsonValues)).
WithFloatVectorColumn(membershipVectorField, membershipVectorDim, vectors))
require.NoError(t, err)
flushLoadMembership(t, ctx, mc, collectionName)
blob, err := client.NewBloomFilterBlob([]int64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}, 0.001)
require.NoError(t, err)
// nested path: meta["a"]["b"] — int-encoded rows (i%3==0) match, float-encoded
// rows (i%3==1) never match.
nested := queryBloomIDs(t, ctx, mc, collectionName, `membership_match(meta["a"]["b"], {bf}, type=bloom)`, blob)
expectedNested := make([]int64, 0, totalRows/3+1)
expectedWhole := make([]int64, 0, totalRows/3+1)
for id := 0; id < totalRows; id++ {
switch id % 3 {
case 0:
expectedNested = append(expectedNested, int64(id))
case 2:
expectedWhole = append(expectedWhole, int64(id))
}
}
require.ElementsMatch(t, expectedNested, nested,
"nested JSON membership must return every int-encoded row and no float/object row")
// whole document: meta — bare-scalar int rows (i%3==2) match.
whole := queryBloomIDs(t, ctx, mc, collectionName, `membership_match(meta, {bf}, type=bloom)`, blob)
require.ElementsMatch(t, expectedWhole, whole,
"whole-document membership must return every bare int scalar and no object row")
}
// TestBloomMatchDynamicFieldPath verifies bloom_match over a dynamic field: an
// unknown identifier resolves to a $meta path, strictly typed per row value.
func TestBloomMatchDynamicFieldPath(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("bloom_dynfield", 6)
schema := entity.NewSchema().WithName(collectionName).
WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
WithField(entity.NewField().WithName(membershipVectorField).WithDataType(entity.FieldTypeFloatVector).WithDim(membershipVectorDim)).
WithDynamicFieldEnabled(true)
require.NoError(t, mc.CreateCollection(ctx, client.NewCreateCollectionOption(collectionName, schema).
WithConsistencyLevel(entity.ClStrong)))
t.Cleanup(func() {
cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
require.NoError(t, mc.DropCollection(cleanupCtx, client.NewDropCollectionOption(collectionName)))
})
const totalRows = 1000
ids := make([]int64, totalRows)
dynInts := make([]int64, totalRows)
vectors := make([][]float32, totalRows)
for i := 0; i < totalRows; i++ {
ids[i] = int64(i)
dynInts[i] = int64(i % 10)
v := make([]float32, membershipVectorDim)
v[0] = float32(i)
vectors[i] = v
}
_, err := mc.Insert(ctx, client.NewColumnBasedInsertOption(collectionName).
WithInt64Column("id", ids).
WithInt64Column("uid", dynInts).
WithFloatVectorColumn(membershipVectorField, membershipVectorDim, vectors))
require.NoError(t, err)
flushLoadMembership(t, ctx, mc, collectionName)
blob, err := client.NewBloomFilterBlob([]int64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}, 0.001)
require.NoError(t, err)
// "uid" is not a schema field -> resolves to a $meta dynamic path.
res := queryBloomIDs(t, ctx, mc, collectionName, `membership_match(uid, {bf}, type=bloom)`, blob)
require.NotEmpty(t, res)
// Every member is present, so all rows match.
require.Len(t, res, totalRows)
}
// TestBloomMatchGrowingAndSealedMixed verifies bloom_match evaluates both sealed
// and growing segments in one query. A blob spanning a sealed member (0..4) and a
// growing-only member (500..504) must return rows from both.
func TestBloomMatchGrowingAndSealedMixed(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("bloom_growing", 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRows(t, ctx, mc, collectionName, false)
flushLoadMembership(t, ctx, mc, collectionName)
// A second batch NOT flushed stays in a growing segment; creator ids 500..509
// are disjoint from the sealed batch's 0..49.
const growingN = 200
gids := make([]int64, growingN)
gcreators := make([]int64, growingN)
gvectors := make([][]float32, growingN)
for i := 0; i < growingN; i++ {
gids[i] = int64(membershipTotalRows + i)
gcreators[i] = int64(500 + i%10)
v := make([]float32, membershipVectorDim)
v[0] = float32(membershipTotalRows + i)
gvectors[i] = v
}
_, err := mc.Insert(ctx, client.NewColumnBasedInsertOption(collectionName).
WithInt64Column("id", gids).
WithInt64Column(membershipCreatorField, gcreators).
WithFloatVectorColumn(membershipVectorField, membershipVectorDim, gvectors))
require.NoError(t, err)
blob, err := client.NewBloomFilterBlob([]int64{0, 1, 2, 3, 4, 500, 501, 502, 503, 504}, 0.001)
require.NoError(t, err)
got := queryBloomIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {bf}, type=bloom)", membershipCreatorField), blob)
expected := expectedMembershipIDs(membershipTotalRows, membershipDomain, map[int64]struct{}{
0: {}, 1: {}, 2: {}, 3: {}, 4: {},
})
for i := 0; i < growingN; i++ {
if i%10 < 5 {
expected = append(expected, int64(membershipTotalRows+i))
}
}
requireBloomResult(t, int64IDSet(got), int64IDSet(expected), membershipTotalRows+growingN,
"sealed and growing segments")
}
// TestBloomMatchFalsePositiveRateSanity verifies the measured false-positive
// count over unique disjoint probe values is within a six-sigma binomial upper
// bound derived from the configured fpr.
func TestBloomMatchFalsePositiveRateSanity(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("bloom_fpr", 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRowsWithDomain(t, ctx, mc, collectionName, false, membershipTotalRows)
flushLoadMembership(t, ctx, mc, collectionName)
// Every creator value is unique, giving 1,990 distinct non-member probes
// instead of repeatedly counting the same 40 values.
const configuredFPR = 0.05
members := []int64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
blob, err := client.NewBloomFilterBlob(members, configuredFPR)
require.NoError(t, err)
got := queryBloomIDSet(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {bf}, type=bloom)", membershipCreatorField), blob)
requireBloomFPRBound(t, got, int64IDSet(members), membershipTotalRows, configuredFPR)
}
// TestBloomMatchEmptyBlob verifies an empty membership blob (0 members) matches
// nothing and is not rejected.
func TestBloomMatchEmptyBlob(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("bloom_empty", 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRows(t, ctx, mc, collectionName, false)
flushLoadMembership(t, ctx, mc, collectionName)
blob, err := client.NewBloomFilterBlob([]int64{}, 0.001)
require.NoError(t, err)
got := queryBloomIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {bf}, type=bloom)", membershipCreatorField), blob)
require.Empty(t, got, "empty blob must match nothing")
}
// TestBloomMatchFprOutOfRange verifies the client builder rejects an out-of-range
// fpr rather than building a filter the server would later reject.
func TestBloomMatchFprOutOfRange(t *testing.T) {
_, err := client.NewBloomFilterBlob([]int64{0, 1, 2}, 0.00001)
require.Error(t, err, "fpr below [0.0001, 0.05] must be rejected")
_, err = client.NewBloomFilterBlob([]int64{0, 1, 2}, 0.06)
require.Error(t, err, "fpr above [0.0001, 0.05] must be rejected")
}
File diff suppressed because it is too large. Load diff
@@ -0,0 +1,153 @@
// Licensed to the LF AI & Data foundation under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// L0 coverage for membership-filter rejection inside struct-array element
// predicates through the public Go SDK.
package testcases
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/require"
"github.com/milvus-io/milvus/client/v3/entity"
"github.com/milvus-io/milvus/client/v3/index"
client "github.com/milvus-io/milvus/client/v3/milvusclient"
"github.com/milvus-io/milvus/tests/go_client/common"
hp "github.com/milvus-io/milvus/tests/go_client/testcases/helper"
)
const membershipElementRows = 200
// setupMembershipElementCollection builds a small struct-array collection with a
// doc-level int64 field (doc_int) and an int64 element sub-field, inserts, flushes,
// indexes and loads it. Used to exercise parser-level rejection of bloom_match /
// roaring_match inside element predicates.
func setupMembershipElementCollection(t *testing.T, ctx CtxT, mc MC, namePrefix string) string {
collName := common.GenRandomString(namePrefix, 6)
opt := hp.DefaultStructAElementSchemaOption(collName)
opt.IncludeDocVChar = false
opt.IncludeCategory = false
schema, structSchema := hp.CreateStructAElementSchema(opt)
require.NoError(t, mc.CreateCollection(ctx,
client.NewCreateCollectionOption(collName, schema).WithConsistencyLevel(entity.ClStrong)))
t.Cleanup(func() {
cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
require.NoError(t, mc.DropCollection(cleanupCtx, client.NewDropCollectionOption(collName)))
})
ds := hp.GenerateStructAElementData(membershipElementRows, 0, opt)
insertElemDataset(t, ctx, mc, collName, structSchema, ds, opt)
flushTask, err := mc.Flush(ctx, client.NewFlushOption(collName))
require.NoError(t, err)
require.NoError(t, flushTask.Await(ctx))
_, err = mc.CreateIndex(ctx, client.NewCreateIndexOption(collName, "normal_vector",
index.NewHNSWIndex(entity.COSINE, 16, 200)))
require.NoError(t, err)
_, err = mc.CreateIndex(ctx, client.NewCreateIndexOption(collName, "structA[embedding]",
index.NewHNSWIndex(entity.MaxSimCosine, 16, 200)))
require.NoError(t, err)
loadTask, err := mc.LoadCollection(ctx, client.NewLoadCollectionOption(collName))
require.NoError(t, err)
require.NoError(t, loadTask.Await(ctx))
return collName
}
// TestMembershipFilterRejectedInElementPredicates verifies bloom_match and
// roaring_match are rejected inside element_filter / MATCH_* element predicates,
// while a document-level sibling conjunction stays legal.
func TestMembershipFilterRejectedInElementPredicates(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collName := setupMembershipElementCollection(t, ctx, mc, "memb_elem")
bloomBlob, err := client.NewBloomFilterBlob([]int64{0, 1, 2}, 0.001)
require.NoError(t, err)
roaringBlob, err := client.NewRoaringBitmapBlob([]int64{0, 1, 2})
require.NoError(t, err)
query := func(expr string, tmplKey string, tmplVal any) error {
opt := client.NewQueryOption(collName).
WithFilter(expr).WithOutputFields("id").
WithConsistencyLevel(entity.ClStrong)
if tmplKey != "" {
opt.WithTemplateParam(tmplKey, tmplVal)
}
_, err := mc.Query(ctx, opt)
return err
}
t.Run("bloom membership inside element_filter rejected", func(t *testing.T) {
err := query(`element_filter(structA, membership_match(doc_int, {bf}, type=bloom) && $[int_val] > 0)`, "bf", bloomBlob)
common.CheckErr(t, err, false, "membership_match filters are not supported inside element_filter")
})
t.Run("roaring membership inside element_filter rejected", func(t *testing.T) {
err := query(`element_filter(structA, membership_match(doc_int, {rb}, type=roaring) && $[int_val] > 0)`, "rb", roaringBlob)
common.CheckErr(t, err, false, "membership_match filters are not supported inside element_filter")
})
t.Run("bloom_match inside MATCH_ANY rejected", func(t *testing.T) {
err := query(`MATCH_ANY(structA, membership_match(doc_int, {bf}, type=bloom) && $[int_val] > 0)`, "bf", bloomBlob)
common.CheckErr(t, err, false, "function calls are not supported inside MATCH predicate")
})
t.Run("document-level sibling conjunction stays legal", func(t *testing.T) {
expr := "membership_match(doc_int, {bf}, type=bloom) && element_filter(structA, $[int_val] > 0)"
got := queryMembershipIDs(t, ctx, mc, collName, expr, "bf", bloomBlob)
gotSet := int64IDSet(got)
require.Len(t, gotSet, len(got), "sibling conjunction returned duplicate PKs")
expected := map[int64]struct{}{0: {}, 1: {}, 2: {}}
requireBloomResult(t, gotSet, expected, membershipElementRows, "document-level sibling conjunction")
})
t.Run("bloom_match on struct sub-field rejected", func(t *testing.T) {
// A struct-array sub-field is an ARRAY type, which bloom_match does not
// support (only INT8/16/32/64/VARCHAR/JSON paths).
err := query(`membership_match(structA[int_val], {bf}, type=bloom)`, "bf", bloomBlob)
common.CheckErr(t, err, false, "membership_match only supports INT8/INT16/INT32/INT64/VARCHAR fields and JSON paths")
})
t.Run("roaring_match on struct sub-field rejected", func(t *testing.T) {
err := query(`membership_match(structA[int_val], {rb}, type=roaring)`, "rb", roaringBlob)
common.CheckErr(t, err, false, "membership_match only supports INT8/INT16/INT32/INT64 fields")
})
t.Run("bloom_match on struct varchar sub-field rejected", func(t *testing.T) {
// A VARCHAR-typed struct sub-field is still an ARRAY to the parser.
strBlob, err := client.NewBloomFilterBlob([]string{"a", "b"}, 0.001)
require.NoError(t, err)
err = query(`membership_match(structA[str_val], {bf}, type=bloom)`, "bf", strBlob)
common.CheckErr(t, err, false, "membership_match only supports INT8/INT16/INT32/INT64/VARCHAR fields and JSON paths")
})
t.Run("bloom_match on struct float sub-field rejected", func(t *testing.T) {
err := query(`membership_match(structA[float_val], {bf}, type=bloom)`, "bf", bloomBlob)
common.CheckErr(t, err, false, "membership_match only supports INT8/INT16/INT32/INT64/VARCHAR fields and JSON paths")
})
t.Run("roaring_match on struct varchar sub-field rejected", func(t *testing.T) {
err := query(`membership_match(structA[str_val], {rb}, type=roaring)`, "rb", roaringBlob)
common.CheckErr(t, err, false, "membership_match only supports INT8/INT16/INT32/INT64 fields")
})
}
@@ -0,0 +1,261 @@
// Licensed to the LF AI & Data foundation under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// Shared int64 fixture for both membership_match kinds: a
// primary-key id, an int64 creator_id field and a small float vector. The row
// count is kept above indexCoord.segment.minSegmentNumRowsToEnableIndex (1024)
// so scalar indexes are really built, not fake-finished.
package testcases
import (
"context"
"sort"
"testing"
"time"
"github.com/stretchr/testify/require"
"github.com/milvus-io/milvus/client/v3/column"
"github.com/milvus-io/milvus/client/v3/entity"
"github.com/milvus-io/milvus/client/v3/index"
client "github.com/milvus-io/milvus/client/v3/milvusclient"
)
const (
membershipCreatorField = "creator_id"
membershipVectorField = "vector"
membershipVectorDim = 8
membershipTotalRows = 2000
membershipDomain = 50
)
func intMembershipSchema(collectionName string, nullable bool) *entity.Schema {
creatorField := entity.NewField().WithName(membershipCreatorField).WithDataType(entity.FieldTypeInt64)
if nullable {
creatorField.WithNullable(true)
}
return entity.NewSchema().WithName(collectionName).
WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
WithField(creatorField).
WithField(entity.NewField().WithName(membershipVectorField).WithDataType(entity.FieldTypeFloatVector).WithDim(membershipVectorDim))
}
func createIntMembershipCollection(t *testing.T, ctx CtxT, mc MC, collectionName string, nullable bool) {
t.Helper()
require.NoError(t, mc.CreateCollection(ctx, client.NewCreateCollectionOption(collectionName, intMembershipSchema(collectionName, nullable)).
WithConsistencyLevel(entity.ClStrong)))
t.Cleanup(func() {
cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
require.NoError(t, mc.DropCollection(cleanupCtx, client.NewDropCollectionOption(collectionName)))
})
}
func insertIntMembershipRows(t *testing.T, ctx CtxT, mc MC, collectionName string, nullable bool) {
insertIntMembershipRowsWithDomain(t, ctx, mc, collectionName, nullable, membershipDomain)
}
func insertIntMembershipRowsWithDomain(
t *testing.T,
ctx CtxT,
mc MC,
collectionName string,
nullable bool,
creatorDomain int,
) {
t.Helper()
require.Positive(t, creatorDomain)
ids := make([]int64, membershipTotalRows)
vectors := make([][]float32, membershipTotalRows)
for i := 0; i < membershipTotalRows; i++ {
ids[i] = int64(i)
v := make([]float32, membershipVectorDim)
v[0] = float32(i)
vectors[i] = v
}
opt := client.NewColumnBasedInsertOption(collectionName).
WithInt64Column("id", ids).
WithFloatVectorColumn(membershipVectorField, membershipVectorDim, vectors)
if nullable {
values := make([]int64, 0, membershipTotalRows)
valid := make([]bool, membershipTotalRows)
for i := 0; i < membershipTotalRows; i++ {
if i%8 == 7 {
valid[i] = false
continue
}
valid[i] = true
values = append(values, int64(i%creatorDomain))
}
col, err := column.NewNullableColumnInt64(membershipCreatorField, values, valid)
require.NoError(t, err)
opt.WithColumns(col)
} else {
creators := make([]int64, membershipTotalRows)
for i := 0; i < membershipTotalRows; i++ {
creators[i] = int64(i % creatorDomain)
}
opt.WithInt64Column(membershipCreatorField, creators)
}
_, err := mc.Insert(ctx, opt)
require.NoError(t, err)
}
func flushLoadMembership(t *testing.T, ctx CtxT, mc MC, collectionName string) {
t.Helper()
flushTask, err := mc.Flush(ctx, client.NewFlushOption(collectionName))
require.NoError(t, err)
require.NoError(t, flushTask.Await(ctx))
vecTask, err := mc.CreateIndex(ctx, client.NewCreateIndexOption(collectionName, membershipVectorField,
index.NewFlatIndex(entity.L2)))
require.NoError(t, err)
require.NoError(t, vecTask.Await(ctx))
loadTask, err := mc.LoadCollection(ctx, client.NewLoadCollectionOption(collectionName))
require.NoError(t, err)
require.NoError(t, loadTask.Await(ctx))
}
// queryMembershipIDs runs a filter query and returns the sorted primary keys.
// The template key differs per filter blob ("bf" or "rb"), so it is passed
// explicitly.
func queryMembershipIDs(t *testing.T, ctx CtxT, mc MC, collectionName, expr, templateKey string, blob any) []int64 {
t.Helper()
opt := client.NewQueryOption(collectionName).
WithFilter(expr).WithOutputFields("id").
WithConsistencyLevel(entity.ClStrong)
if blob != nil {
opt.WithTemplateParam(templateKey, blob)
}
rs, err := mc.Query(ctx, opt)
require.NoError(t, err, "query %q", expr)
col, ok := rs.GetColumn("id").(*column.ColumnInt64)
require.True(t, ok)
out := make([]int64, 0, col.Len())
for i := 0; i < col.Len(); i++ {
v, err := col.GetAsInt64(i)
require.NoError(t, err)
out = append(out, v)
}
sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
return out
}
func queryMembershipIDSet(t *testing.T, ctx CtxT, mc MC, collectionName, expr, templateKey string, blob any) map[int64]struct{} {
t.Helper()
m := make(map[int64]struct{})
for _, id := range queryMembershipIDs(t, ctx, mc, collectionName, expr, templateKey, blob) {
m[id] = struct{}{}
}
return m
}
func expectedMembershipIDs(totalRows, domain int, members map[int64]struct{}) []int64 {
ids := make([]int64, 0, totalRows)
for i := 0; i < totalRows; i++ {
if _, ok := members[int64(i%domain)]; ok {
ids = append(ids, int64(i))
}
}
return ids
}
func int64IDSet(values []int64) map[int64]struct{} {
result := make(map[int64]struct{}, len(values))
for _, value := range values {
result[value] = struct{}{}
}
return result
}
// requireBloomResult verifies both sides of Bloom's approximate contract:
// every true member must be present, while false positives stay far below an
// ignored-filter/full-scan result. The 30% ceiling is deliberately much looser
// than the builders' configured 0.1%/5% FPR to avoid statistical flakiness.
func requireBloomResult(
t *testing.T,
got map[int64]struct{},
expected map[int64]struct{},
universeSize int,
context string,
) {
t.Helper()
require.LessOrEqual(t, len(expected), universeSize, "invalid Bloom ground truth: %s", context)
for pk := range expected {
require.Containsf(t, got, pk, "Bloom result dropped true member id=%d: %s", pk, context)
}
falsePositives := len(got) - len(expected)
nonMembers := universeSize - len(expected)
if nonMembers == 0 {
require.Zero(t, falsePositives, "all rows are true members: %s", context)
return
}
require.Lessf(t, falsePositives*10, nonMembers*3,
"Bloom result has %d false positives over %d non-members; filter may be ignored: %s",
falsePositives, nonMembers, context)
}
func searchMembershipIDs(t *testing.T, ctx CtxT, mc MC, collectionName, expr, templateKey string, blob any) []int64 {
t.Helper()
opt := client.NewSearchOption(collectionName, membershipTotalRows,
[]entity.Vector{entity.FloatVector(make([]float32, membershipVectorDim))}).
WithANNSField(membershipVectorField).
WithFilter(expr).
WithOutputFields(membershipCreatorField).
WithConsistencyLevel(entity.ClStrong)
if blob != nil {
opt.WithTemplateParam(templateKey, blob)
}
results, err := mc.Search(ctx, opt)
require.NoError(t, err, "search %q", expr)
require.Len(t, results, 1)
require.NoError(t, results[0].Err)
result := results[0]
pkColumn, ok := result.IDs.(*column.ColumnInt64)
require.True(t, ok, "search result primary key must be INT64")
require.Equal(t, result.ResultCount, pkColumn.Len(), "Search result count and PK count differ")
require.Len(t, result.Scores, result.ResultCount, "Search score count and result count differ")
creatorColumn, ok := result.GetColumn(membershipCreatorField).(*column.ColumnInt64)
require.True(t, ok, "requested creator_id output field must be INT64")
require.Equal(t, result.ResultCount, creatorColumn.Len(), "output field and result counts differ")
ids := make([]int64, 0, result.ResultCount)
for i := 0; i < pkColumn.Len(); i++ {
id, err := pkColumn.GetAsInt64(i)
require.NoError(t, err)
creator, err := creatorColumn.GetAsInt64(i)
require.NoError(t, err)
require.Equal(t, id%membershipDomain, creator, "creator_id projection does not match PK %d", id)
if i > 0 {
require.LessOrEqual(t, result.Scores[i-1], result.Scores[i], "L2 Search scores must be nondecreasing")
}
ids = append(ids, id)
}
sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
return ids
}
@@ -0,0 +1,109 @@
// Licensed to the LF AI & Data foundation under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package testcases
import (
"fmt"
"testing"
"time"
"github.com/stretchr/testify/require"
"github.com/milvus-io/milvus/client/v3/entity"
client "github.com/milvus-io/milvus/client/v3/milvusclient"
"github.com/milvus-io/milvus/tests/go_client/common"
hp "github.com/milvus-io/milvus/tests/go_client/testcases/helper"
)
// TestMembershipMatchUnifiedContract pins the public contract introduced by
// #52778: magic-based dispatch is equivalent to an explicit type pin, a pin
// must agree with the blob magic, unknown magic fails closed, and the two
// unreleased predecessor function names remain rejected.
func TestMembershipMatchUnifiedContract(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("membership_contract", 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRows(t, ctx, mc, collectionName, false)
flushLoadMembership(t, ctx, mc, collectionName)
bloomBlob, err := client.NewBloomFilterBlob([]int64{0, 1, 2, 42}, 0.001)
require.NoError(t, err)
roaringBlob, err := client.NewRoaringBitmapBlob([]int64{0, 1, 2, 42})
require.NoError(t, err)
t.Run("auto dispatch matches explicit type", func(t *testing.T) {
bloomAuto := queryMembershipIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {bf})", membershipCreatorField), "bf", bloomBlob)
bloomPinned := queryMembershipIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {bf}, type=bloom)", membershipCreatorField), "bf", bloomBlob)
expectedBloom := expectedMembershipIDs(membershipTotalRows, membershipDomain,
map[int64]struct{}{0: {}, 1: {}, 2: {}, 42: {}})
requireBloomResult(t, int64IDSet(bloomAuto), int64IDSet(expectedBloom), membershipTotalRows,
"auto-dispatched Bloom")
requireBloomResult(t, int64IDSet(bloomPinned), int64IDSet(expectedBloom), membershipTotalRows,
"explicit Bloom")
require.ElementsMatch(t, bloomPinned, bloomAuto)
roaringAuto := queryMembershipIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {rb})", membershipCreatorField), "rb", roaringBlob)
roaringPinned := queryMembershipIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {rb}, type=roaring)", membershipCreatorField), "rb", roaringBlob)
require.ElementsMatch(t, roaringPinned, roaringAuto)
})
queryErr := func(expr, key string, blob any) error {
opt := client.NewQueryOption(collectionName).
WithFilter(expr).
WithOutputFields("id").
WithConsistencyLevel(entity.ClStrong).
WithTemplateParam(key, blob)
_, err := mc.Query(ctx, opt)
return err
}
t.Run("explicit type must match magic", func(t *testing.T) {
err := queryErr(fmt.Sprintf("membership_match(%s, {bf}, type=roaring)", membershipCreatorField), "bf", bloomBlob)
require.ErrorContains(t, err, "does not match filter blob format")
err = queryErr(fmt.Sprintf("membership_match(%s, {rb}, type=bloom)", membershipCreatorField), "rb", roaringBlob)
require.ErrorContains(t, err, "does not match filter blob format")
})
t.Run("unknown magic fails closed", func(t *testing.T) {
err := queryErr(fmt.Sprintf("membership_match(%s, {blob})", membershipCreatorField), "blob",
client.BloomFilterBlob([]byte("UNKNOWN-MEMBERSHIP-FORMAT")))
require.ErrorContains(t, err, "unknown format magic")
})
t.Run("predecessor names are rejected", func(t *testing.T) {
cases := []struct {
expr string
key string
blob any
}{
{fmt.Sprintf("bloom_match(%s, {bf})", membershipCreatorField), "bf", bloomBlob},
{fmt.Sprintf("roaring_match(%s, {rb})", membershipCreatorField), "rb", roaringBlob},
}
for _, tc := range cases {
err := queryErr(tc.expr, tc.key, tc.blob)
require.ErrorContains(t, err, "is not supported", tc.expr)
require.ErrorContains(t, err, "membership_match", tc.expr)
}
})
}
@@ -0,0 +1,145 @@
// Licensed to the LF AI & Data foundation under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package testcases
import (
"encoding/hex"
"encoding/json"
"fmt"
"os"
"path/filepath"
"runtime"
"testing"
"time"
"github.com/stretchr/testify/require"
client "github.com/milvus-io/milvus/client/v3/milvusclient"
"github.com/milvus-io/milvus/tests/go_client/common"
hp "github.com/milvus-io/milvus/tests/go_client/testcases/helper"
)
type membershipWireKindFixture struct {
Domain string `json:"domain"`
Members []int64 `json:"members"`
PresentMembers []int64 `json:"present_members"`
FPR float64 `json:"fpr"`
Hex string `json:"hex"`
}
type membershipWireFixture struct {
SchemaVersion int `json:"schema_version"`
Bloom membershipWireKindFixture `json:"bloom"`
Roaring membershipWireKindFixture `json:"roaring"`
}
func loadMembershipWireFixture(t *testing.T) membershipWireFixture {
t.Helper()
path := membershipWireFixturePath(t)
payload, err := os.ReadFile(path)
require.NoError(t, err)
var fixture membershipWireFixture
require.NoError(t, json.Unmarshal(payload, &fixture))
require.Equal(t, 1, fixture.SchemaVersion)
require.Equal(t, "int64", fixture.Bloom.Domain)
require.Equal(t, "int64", fixture.Roaring.Domain)
return fixture
}
func membershipWireFixturePath(t *testing.T) string {
t.Helper()
const manifest = "manifest.json"
candidates := make([]string, 0, 8)
if _, sourceFile, _, ok := runtime.Caller(0); ok && filepath.IsAbs(sourceFile) {
candidates = append(candidates,
filepath.Join(filepath.Dir(sourceFile), "..", "..", "fixtures", "membership_filter", manifest))
}
workingDir, err := os.Getwd()
require.NoError(t, err)
for dir := workingDir; ; dir = filepath.Dir(dir) {
candidates = append(candidates,
filepath.Join(dir, "tests", "fixtures", "membership_filter", manifest),
filepath.Join(dir, "fixtures", "membership_filter", manifest))
parent := filepath.Dir(dir)
if parent == dir {
break
}
}
for _, candidate := range candidates {
if _, err := os.Stat(candidate); err == nil {
return candidate
}
}
t.Fatalf("membership wire fixture not found; searched %v", candidates)
return ""
}
func int64MemberSet(values []int64) map[int64]struct{} {
result := make(map[int64]struct{}, len(values))
for _, value := range values {
result[value] = struct{}{}
}
return result
}
func TestMembershipWireFixtureManifest(t *testing.T) {
fixture := loadMembershipWireFixture(t)
require.NotEmpty(t, fixture.Bloom.Hex)
require.NotEmpty(t, fixture.Roaring.Hex)
}
func TestMembershipPythonGoldenBytes(t *testing.T) {
fixture := loadMembershipWireFixture(t)
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
// Both golden blobs target the same INT64 membership schema and read-only
// sealed fixture, so build/load it once and exercise both wire formats on the
// same collection.
collectionName := common.GenRandomString("membership_wire", 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRows(t, ctx, mc, collectionName, false)
flushLoadMembership(t, ctx, mc, collectionName)
t.Run("bloom", func(t *testing.T) {
pythonBlob, err := hex.DecodeString(fixture.Bloom.Hex)
require.NoError(t, err)
goBlob, err := client.NewBloomFilterBlob(fixture.Bloom.Members, fixture.Bloom.FPR)
require.NoError(t, err)
require.Equal(t, pythonBlob, []byte(goBlob), "Go and pymilvus bloom builders must share the wire format")
got := queryMembershipIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {bf}, type=bloom)", membershipCreatorField), "bf", client.BloomFilterBlob(pythonBlob))
expected := expectedMembershipIDs(membershipTotalRows, membershipDomain, int64MemberSet(fixture.Bloom.PresentMembers))
requireBloomResult(t, int64IDSet(got), int64IDSet(expected), membershipTotalRows,
"pymilvus-built golden Bloom blob")
})
t.Run("roaring", func(t *testing.T) {
pythonBlob, err := hex.DecodeString(fixture.Roaring.Hex)
require.NoError(t, err)
goBlob, err := client.NewRoaringBitmapBlob(fixture.Roaring.Members)
require.NoError(t, err)
require.Equal(t, pythonBlob, []byte(goBlob), "Go and pymilvus roaring builders must share the wire format")
got := queryMembershipIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {rb}, type=roaring)", membershipCreatorField), "rb", client.RoaringBitmapBlob(pythonBlob))
expected := expectedMembershipIDs(membershipTotalRows, membershipDomain, int64MemberSet(fixture.Roaring.PresentMembers))
require.ElementsMatch(t, expected, got, "pymilvus-built roaring blob result mismatch")
})
}
@@ -0,0 +1,673 @@
// Licensed to the LF AI & Data foundation under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// Roaring-kind membership_match coverage through the public Go SDK.
package testcases
import (
"context"
"fmt"
"math"
"testing"
"time"
"github.com/stretchr/testify/require"
"github.com/milvus-io/milvus/client/v3/entity"
"github.com/milvus-io/milvus/client/v3/index"
client "github.com/milvus-io/milvus/client/v3/milvusclient"
"github.com/milvus-io/milvus/tests/go_client/common"
hp "github.com/milvus-io/milvus/tests/go_client/testcases/helper"
)
func queryRoaringIDs(t *testing.T, ctx CtxT, mc MC, collectionName, expr string, blob any) []int64 {
t.Helper()
return queryMembershipIDs(t, ctx, mc, collectionName, expr, "rb", blob)
}
// TestRoaringMatchExactMembership verifies roaring_match selects exactly the
// requested ids that exist — no false positives and no false negatives.
func TestRoaringMatchExactMembership(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("roaring_exact", 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRows(t, ctx, mc, collectionName, false)
flushLoadMembership(t, ctx, mc, collectionName)
members := []int64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1000000} // last one absent
blob, err := client.NewRoaringBitmapBlob(members)
require.NoError(t, err)
got := queryRoaringIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {rb}, type=roaring)", membershipCreatorField), blob)
expected := expectedMembershipIDs(membershipTotalRows, membershipDomain, map[int64]struct{}{
0: {}, 1: {}, 2: {}, 3: {}, 4: {}, 5: {}, 6: {}, 7: {}, 8: {}, 9: {},
})
require.ElementsMatch(t, expected, got, "roaring_match must return the exact matching row PK set")
searchGot := searchMembershipIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {rb}, type=roaring)", membershipCreatorField), "rb", blob)
require.ElementsMatch(t, expected, searchGot, "roaring_match Search must return the exact row PK set")
}
// TestRoaringMatchNotExactComplement verifies `not roaring_match` returns the
// exact complement over non-NULL rows.
func TestRoaringMatchNotExactComplement(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("roaring_not", 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRows(t, ctx, mc, collectionName, false)
flushLoadMembership(t, ctx, mc, collectionName)
blob, err := client.NewRoaringBitmapBlob([]int64{0, 1, 2, 3, 4})
require.NoError(t, err)
got := queryRoaringIDs(t, ctx, mc, collectionName,
fmt.Sprintf("not membership_match(%s, {rb}, type=roaring)", membershipCreatorField), blob)
complement := make(map[int64]struct{}, membershipDomain-5)
for value := int64(5); value < membershipDomain; value++ {
complement[value] = struct{}{}
}
expected := expectedMembershipIDs(membershipTotalRows, membershipDomain, complement)
require.ElementsMatch(t, expected, got, "not roaring_match must return the exact complement row PK set")
}
// TestRoaringMatchDelete verifies roaring_match is accepted in a delete
// expression (bloom_match is not, since it has false positives) and removes
// exactly the requested ids.
func TestRoaringMatchDelete(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("roaring_delete", 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRows(t, ctx, mc, collectionName, false)
flushLoadMembership(t, ctx, mc, collectionName)
victims := []int64{0, 1, 2}
blob, err := client.NewRoaringBitmapBlob(victims)
require.NoError(t, err)
dr, err := mc.Delete(ctx, client.NewDeleteOption(collectionName).
WithExpr(fmt.Sprintf("membership_match(%s, {rb}, type=roaring)", membershipCreatorField)).
WithTemplateParam("rb", blob))
require.NoError(t, err)
// Fixture couples pk id == row index i and creator == i % domain, so each
// creator value 0..domain-1 appears exactly totalRows/domain times; deleting
// len(victims) distinct values removes len(victims) * totalRows/domain rows.
require.EqualValues(t, int64(len(victims)*membershipTotalRows/membershipDomain), dr.DeleteCount)
remaining := queryRoaringIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {rb}, type=roaring)", membershipCreatorField), blob)
require.Empty(t, remaining, "victim rows must be deleted")
}
// TestRoaringMatchGrowingAndSealedMixed verifies roaring_match evaluates both
// sealed and growing segments in one query, exactly.
func TestRoaringMatchGrowingAndSealedMixed(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("roaring_growing", 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRows(t, ctx, mc, collectionName, false)
flushLoadMembership(t, ctx, mc, collectionName)
const growingN = 200
gids := make([]int64, growingN)
gcreators := make([]int64, growingN)
gvectors := make([][]float32, growingN)
for i := 0; i < growingN; i++ {
gids[i] = int64(membershipTotalRows + i)
gcreators[i] = int64(500 + i%10)
v := make([]float32, membershipVectorDim)
v[0] = float32(membershipTotalRows + i)
gvectors[i] = v
}
_, err := mc.Insert(ctx, client.NewColumnBasedInsertOption(collectionName).
WithInt64Column("id", gids).
WithInt64Column(membershipCreatorField, gcreators).
WithFloatVectorColumn(membershipVectorField, membershipVectorDim, gvectors))
require.NoError(t, err)
blob, err := client.NewRoaringBitmapBlob([]int64{0, 1, 2, 3, 4, 500, 501, 502, 503, 504})
require.NoError(t, err)
got := queryRoaringIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {rb}, type=roaring)", membershipCreatorField), blob)
expected := expectedMembershipIDs(membershipTotalRows, membershipDomain, map[int64]struct{}{
0: {}, 1: {}, 2: {}, 3: {}, 4: {},
})
for i := 0; i < growingN; i++ {
if i%10 < 5 {
expected = append(expected, int64(membershipTotalRows+i))
}
}
require.ElementsMatch(t, expected, got,
"roaring_match must return every exact member across sealed and growing segments")
}
// TestRoaringMatchEmptyBitmap verifies an empty bitmap matches nothing.
func TestRoaringMatchEmptyBitmap(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("roaring_empty", 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRows(t, ctx, mc, collectionName, false)
flushLoadMembership(t, ctx, mc, collectionName)
blob, err := client.NewRoaringBitmapBlob([]int64{})
require.NoError(t, err)
got := queryRoaringIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {rb}, type=roaring)", membershipCreatorField), blob)
require.Empty(t, got, "empty bitmap must match nothing")
}
// TestRoaringMatchInvalidInputRejected verifies a literal array, malformed blob
// and truncated blob are rejected rather than silently unfiltered.
func TestRoaringMatchInvalidInputRejected(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("roaring_reject", 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRows(t, ctx, mc, collectionName, false)
flushLoadMembership(t, ctx, mc, collectionName)
// literal array argument
_, err := mc.Query(ctx, client.NewQueryOption(collectionName).
WithFilter(fmt.Sprintf("membership_match(%s, [1,2,3], type=roaring)", membershipCreatorField)).WithOutputFields("id").
WithConsistencyLevel(entity.ClStrong))
common.CheckErr(t, err, false, "must be a {template} placeholder")
// malformed MRB1 blob
_, err = mc.Query(ctx, client.NewQueryOption(collectionName).
WithFilter(fmt.Sprintf("membership_match(%s, {rb}, type=roaring)", membershipCreatorField)).WithOutputFields("id").
WithTemplateParam("rb", client.RoaringBitmapBlob([]byte("not-an-mrb1-blob"))).
WithConsistencyLevel(entity.ClStrong))
common.CheckErr(t, err, false, "unknown format magic")
// truncated valid blob
valid, err := client.NewRoaringBitmapBlob([]int64{1, 2, 3})
require.NoError(t, err)
_, err = mc.Query(ctx, client.NewQueryOption(collectionName).
WithFilter(fmt.Sprintf("membership_match(%s, {rb}, type=roaring)", membershipCreatorField)).WithOutputFields("id").
WithTemplateParam("rb", valid[:len(valid)-1]).
WithConsistencyLevel(entity.ClStrong))
common.CheckErr(t, err, false, "membership_match roaring bitmap blob is invalid")
}
// TestRoaringMatchNullRowsFoldToFalse verifies NULL rows never match
// roaring_match nor its negation.
func TestRoaringMatchNullRowsFoldToFalse(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("roaring_null", 6)
createIntMembershipCollection(t, ctx, mc, collectionName, true)
insertIntMembershipRows(t, ctx, mc, collectionName, true)
flushLoadMembership(t, ctx, mc, collectionName)
members := make([]int64, 0, membershipDomain)
for v := 0; v < membershipDomain; v++ {
members = append(members, int64(v))
}
blob, err := client.NewRoaringBitmapBlob(members)
require.NoError(t, err)
got := queryRoaringIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {rb}, type=roaring)", membershipCreatorField), blob)
expected := make([]int64, 0, membershipTotalRows-membershipTotalRows/8)
for id := 0; id < membershipTotalRows; id++ {
if id%8 != 7 {
expected = append(expected, int64(id))
}
}
require.ElementsMatch(t, expected, got, "all and only non-NULL rows must match when every value is a member")
notRs := queryRoaringIDs(t, ctx, mc, collectionName,
fmt.Sprintf("not membership_match(%s, {rb}, type=roaring)", membershipCreatorField), blob)
require.Empty(t, notRs, "NULL folds to false and every non-NULL value matches, so negation must match nothing")
}
// TestRoaringMatchNegativeValues verifies negative ids round-trip through the
// bitmap's two's-complement encoding. Negative ids map to the top of the uint64
// key space and land in a different Roaring high container than positive ids; a
// build/probe pair that zero-extended instead of sign-extending would still pass
// every all-positive test, so this pins the highest-risk case.
func TestRoaringMatchNegativeValues(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("roaring_negative", 6)
const (
domain = 100
negShift = 50 // creator = i % domain - negShift, straddling zero
totalRows = 2000
)
schema := entity.NewSchema().WithName(collectionName).
WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
WithField(entity.NewField().WithName(membershipCreatorField).WithDataType(entity.FieldTypeInt64)).
WithField(entity.NewField().WithName(membershipVectorField).WithDataType(entity.FieldTypeFloatVector).WithDim(membershipVectorDim))
require.NoError(t, mc.CreateCollection(ctx, client.NewCreateCollectionOption(collectionName, schema).
WithConsistencyLevel(entity.ClStrong)))
t.Cleanup(func() {
cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
require.NoError(t, mc.DropCollection(cleanupCtx, client.NewDropCollectionOption(collectionName)))
})
ids := make([]int64, totalRows)
creators := make([]int64, totalRows)
vectors := make([][]float32, totalRows)
for i := 0; i < totalRows; i++ {
ids[i] = int64(i)
creators[i] = int64(i%domain) - negShift
v := make([]float32, membershipVectorDim)
v[0] = float32(i)
vectors[i] = v
}
_, err := mc.Insert(ctx, client.NewColumnBasedInsertOption(collectionName).
WithInt64Column("id", ids).
WithInt64Column(membershipCreatorField, creators).
WithFloatVectorColumn(membershipVectorField, membershipVectorDim, vectors))
require.NoError(t, err)
flushLoadMembership(t, ctx, mc, collectionName)
// Half negative, half positive, plus values absent from the collection.
members := []int64{-negShift, -30, -1, 0, 1, 17, 49, 1 << 20, -(1 << 20)}
blob, err := client.NewRoaringBitmapBlob(members)
require.NoError(t, err)
got := queryRoaringIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {rb}, type=roaring)", membershipCreatorField), blob)
expected := make([]int64, 0, totalRows)
for id, creator := range creators {
if isMember(members, creator) {
expected = append(expected, int64(id))
}
}
require.NotEmpty(t, expected, "test setup must select at least one present id")
require.ElementsMatch(t, expected, got,
"negative and positive members must return every matching row, not merely one row per value")
}
// TestRoaringMatchScalarIndexTypeMatrix verifies roaring_match stays exact when
// a scalar index is built on the field before load. STL_SORT drops the raw
// column (has_raw_data=true), forcing roaring_match through the index's
// Reverse_Lookup path; INVERTED and BITMAP cover the other supported scalar
// index entry points. All must select exactly the requested ids.
func TestRoaringMatchScalarIndexTypeMatrix(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
members := []int64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1000000}
blob, err := client.NewRoaringBitmapBlob(members)
require.NoError(t, err)
for _, indexType := range []string{"STL_SORT", "INVERTED", "BITMAP"} {
t.Run(indexType, func(t *testing.T) {
collectionName := common.GenRandomString("roaring_idx_"+indexType, 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRows(t, ctx, mc, collectionName, false)
flushTask, err := mc.Flush(ctx, client.NewFlushOption(collectionName))
require.NoError(t, err)
require.NoError(t, flushTask.Await(ctx))
var scalarIndex index.Index
switch indexType {
case "STL_SORT":
scalarIndex = index.NewSortedIndex()
case "INVERTED":
scalarIndex = index.NewInvertedIndex()
case "BITMAP":
scalarIndex = index.NewBitmapIndex()
}
idxTask, err := mc.CreateIndex(ctx, client.NewCreateIndexOption(collectionName, membershipCreatorField,
scalarIndex))
require.NoError(t, err)
require.NoError(t, idxTask.Await(ctx))
vecTask, err := mc.CreateIndex(ctx, client.NewCreateIndexOption(collectionName, membershipVectorField,
index.NewFlatIndex(entity.L2)))
require.NoError(t, err)
require.NoError(t, vecTask.Await(ctx))
loadTask, err := mc.LoadCollection(ctx, client.NewLoadCollectionOption(collectionName))
require.NoError(t, err)
require.NoError(t, loadTask.Await(ctx))
got := queryRoaringIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {rb}, type=roaring)", membershipCreatorField), blob)
expected := expectedMembershipIDs(membershipTotalRows, membershipDomain, map[int64]struct{}{
0: {}, 1: {}, 2: {}, 3: {}, 4: {}, 5: {}, 6: {}, 7: {}, 8: {}, 9: {},
})
require.ElementsMatch(t, expected, got, "roaring_match row set mismatch under %s", indexType)
})
}
}
// TestRoaringMatchAutoIndex verifies roaring_match stays exact when the field is
// indexed with AUTOINDEX for both low- and high-cardinality datasets, without
// coupling the assertion to AUTOINDEX's internal implementation choice.
func TestRoaringMatchAutoIndex(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
members := []int64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1000000}
blob, err := client.NewRoaringBitmapBlob(members)
require.NoError(t, err)
// Exercise AUTOINDEX with both high- and low-cardinality inputs. Explicit
// index-path coverage lives in TestRoaringMatchScalarIndexTypeMatrix.
for _, domain := range []int{200, 50} {
t.Run(fmt.Sprintf("domain_%d", domain), func(t *testing.T) {
collectionName := common.GenRandomString("roaring_auto", 6)
schema := entity.NewSchema().WithName(collectionName).
WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
WithField(entity.NewField().WithName(membershipCreatorField).WithDataType(entity.FieldTypeInt64)).
WithField(entity.NewField().WithName(membershipVectorField).WithDataType(entity.FieldTypeFloatVector).WithDim(membershipVectorDim))
require.NoError(t, mc.CreateCollection(ctx, client.NewCreateCollectionOption(collectionName, schema).
WithConsistencyLevel(entity.ClStrong)))
t.Cleanup(func() {
cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
require.NoError(t, mc.DropCollection(cleanupCtx, client.NewDropCollectionOption(collectionName)))
})
ids := make([]int64, membershipTotalRows)
creators := make([]int64, membershipTotalRows)
vectors := make([][]float32, membershipTotalRows)
for i := 0; i < membershipTotalRows; i++ {
ids[i] = int64(i)
creators[i] = int64(i % domain)
v := make([]float32, membershipVectorDim)
v[0] = float32(i)
vectors[i] = v
}
_, err := mc.Insert(ctx, client.NewColumnBasedInsertOption(collectionName).
WithInt64Column("id", ids).
WithInt64Column(membershipCreatorField, creators).
WithFloatVectorColumn(membershipVectorField, membershipVectorDim, vectors))
require.NoError(t, err)
flushTask, err := mc.Flush(ctx, client.NewFlushOption(collectionName))
require.NoError(t, err)
require.NoError(t, flushTask.Await(ctx))
idxTask, err := mc.CreateIndex(ctx, client.NewCreateIndexOption(collectionName, membershipCreatorField,
index.NewAutoIndex(entity.L2)))
require.NoError(t, err)
require.NoError(t, idxTask.Await(ctx))
vecTask, err := mc.CreateIndex(ctx, client.NewCreateIndexOption(collectionName, membershipVectorField,
index.NewFlatIndex(entity.L2)))
require.NoError(t, err)
require.NoError(t, vecTask.Await(ctx))
loadTask, err := mc.LoadCollection(ctx, client.NewLoadCollectionOption(collectionName))
require.NoError(t, err)
require.NoError(t, loadTask.Await(ctx))
got := queryRoaringIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {rb}, type=roaring)", membershipCreatorField), blob)
expected := expectedMembershipIDs(membershipTotalRows, domain, map[int64]struct{}{
0: {}, 1: {}, 2: {}, 3: {}, 4: {}, 5: {}, 6: {}, 7: {}, 8: {}, 9: {},
})
require.ElementsMatch(t, expected, got, "roaring_match row set mismatch under AUTOINDEX domain %d", domain)
})
}
}
// TestRoaringMatchIntTypeMatrix verifies one int64-built bitmap probes every
// integer field width, including the sign-extension of narrow negative values.
func TestRoaringMatchIntTypeMatrix(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("roaring_intmatrix", 6)
const (
domain = 100
shift = 50 // value = i % domain - shift, straddling zero
)
schema := entity.NewSchema().WithName(collectionName).
WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
WithField(entity.NewField().WithName("i8").WithDataType(entity.FieldTypeInt8)).
WithField(entity.NewField().WithName("i16").WithDataType(entity.FieldTypeInt16)).
WithField(entity.NewField().WithName("i32").WithDataType(entity.FieldTypeInt32)).
WithField(entity.NewField().WithName("i64").WithDataType(entity.FieldTypeInt64)).
WithField(entity.NewField().WithName(membershipVectorField).WithDataType(entity.FieldTypeFloatVector).WithDim(membershipVectorDim))
require.NoError(t, mc.CreateCollection(ctx, client.NewCreateCollectionOption(collectionName, schema).
WithConsistencyLevel(entity.ClStrong)))
t.Cleanup(func() {
cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
require.NoError(t, mc.DropCollection(cleanupCtx, client.NewDropCollectionOption(collectionName)))
})
const totalRows = 2000
ids := make([]int64, totalRows)
i8 := make([]int8, totalRows)
i16 := make([]int16, totalRows)
i32 := make([]int32, totalRows)
i64 := make([]int64, totalRows)
vectors := make([][]float32, totalRows)
for i := 0; i < totalRows; i++ {
ids[i] = int64(i)
v := int64(i%domain) - shift
i8[i] = int8(v)
i16[i] = int16(v)
i32[i] = int32(v)
i64[i] = v
vec := make([]float32, membershipVectorDim)
vec[0] = float32(i)
vectors[i] = vec
}
_, err := mc.Insert(ctx, client.NewColumnBasedInsertOption(collectionName).
WithInt64Column("id", ids).
WithInt8Column("i8", i8).
WithInt16Column("i16", i16).
WithInt32Column("i32", i32).
WithInt64Column("i64", i64).
WithFloatVectorColumn(membershipVectorField, membershipVectorDim, vectors))
require.NoError(t, err)
flushLoadMembership(t, ctx, mc, collectionName)
// Straddle zero so a zero-extending narrow build/probe would place a negative
// value in the wrong Roaring high container and miss.
members := []int64{-shift, -30, -1, 0, 1, 17, 49}
blob, err := client.NewRoaringBitmapBlob(members)
require.NoError(t, err)
for _, field := range []string{"i8", "i16", "i32", "i64"} {
got := queryRoaringIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {rb}, type=roaring)", field), blob)
expected := make([]int64, 0, totalRows)
for id := 0; id < totalRows; id++ {
if isMember(members, int64(id%domain-shift)) {
expected = append(expected, int64(id))
}
}
require.ElementsMatchf(t, expected, got,
"roaring_match must return every matching PK across integer widths on %s", field)
}
}
func isMember(members []int64, v int64) bool {
for _, m := range members {
if m == v {
return true
}
}
return false
}
// TestRoaringMatchInt64Bounds verifies INT64_MIN/MAX round-trip through the
// two's-complement mapping.
func TestRoaringMatchInt64Bounds(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("roaring_bounds", 6)
schema := entity.NewSchema().WithName(collectionName).
WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true)).
WithField(entity.NewField().WithName(membershipCreatorField).WithDataType(entity.FieldTypeInt64)).
WithField(entity.NewField().WithName(membershipVectorField).WithDataType(entity.FieldTypeFloatVector).WithDim(membershipVectorDim))
require.NoError(t, mc.CreateCollection(ctx, client.NewCreateCollectionOption(collectionName, schema).
WithConsistencyLevel(entity.ClStrong)))
t.Cleanup(func() {
cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
require.NoError(t, mc.DropCollection(cleanupCtx, client.NewDropCollectionOption(collectionName)))
})
const totalRows = 2000
ids := make([]int64, totalRows)
creators := make([]int64, totalRows)
vectors := make([][]float32, totalRows)
for i := 0; i < totalRows; i++ {
ids[i] = int64(i)
v := make([]float32, membershipVectorDim)
v[0] = float32(i)
vectors[i] = v
switch i % 4 {
case 0:
creators[i] = math.MinInt64
case 1:
creators[i] = math.MaxInt64
case 2:
creators[i] = -1
default:
creators[i] = 42
}
}
_, err := mc.Insert(ctx, client.NewColumnBasedInsertOption(collectionName).
WithInt64Column("id", ids).
WithInt64Column(membershipCreatorField, creators).
WithFloatVectorColumn(membershipVectorField, membershipVectorDim, vectors))
require.NoError(t, err)
flushLoadMembership(t, ctx, mc, collectionName)
blob, err := client.NewRoaringBitmapBlob([]int64{math.MinInt64, math.MaxInt64, -1, 42})
require.NoError(t, err)
got := queryRoaringIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {rb}, type=roaring)", membershipCreatorField), blob)
require.Len(t, got, totalRows, "all four values are present, so every row must match")
// A single absent extreme must match nothing.
absent, err := client.NewRoaringBitmapBlob([]int64{math.MaxInt64 - 1})
require.NoError(t, err)
gotAbsent := queryRoaringIDs(t, ctx, mc, collectionName,
fmt.Sprintf("membership_match(%s, {rb}, type=roaring)", membershipCreatorField), absent)
require.Empty(t, gotAbsent)
}
// TestRoaringMatchNotEmptyBitmap verifies `not membership_match(field, {empty})`
// selects every non-NULL row (the exact complement of an empty set).
func TestRoaringMatchNotEmptyBitmap(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
collectionName := common.GenRandomString("roaring_notempty", 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRows(t, ctx, mc, collectionName, false)
flushLoadMembership(t, ctx, mc, collectionName)
blob, err := client.NewRoaringBitmapBlob([]int64{})
require.NoError(t, err)
got := queryRoaringIDs(t, ctx, mc, collectionName,
fmt.Sprintf("not membership_match(%s, {rb}, type=roaring)", membershipCreatorField), blob)
require.Len(t, got, membershipTotalRows, "negating an empty bitmap must select every non-NULL row")
}
// TestRoaringMatchDeleteThreeStates verifies positive-empty, negated-nonempty,
// and negated-empty deletion (delete nothing, delete the complement, delete all).
func TestRoaringMatchDeleteThreeStates(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
mc := hp.CreateDefaultMilvusClient(ctx, t)
// Positive empty: delete nothing.
t.Run("positive_empty", func(t *testing.T) {
collectionName := common.GenRandomString("roaring_del_pos_empty", 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRows(t, ctx, mc, collectionName, false)
flushLoadMembership(t, ctx, mc, collectionName)
blob, err := client.NewRoaringBitmapBlob([]int64{})
require.NoError(t, err)
dr, err := mc.Delete(ctx, client.NewDeleteOption(collectionName).
WithExpr(fmt.Sprintf("membership_match(%s, {rb}, type=roaring)", membershipCreatorField)).
WithTemplateParam("rb", blob))
require.NoError(t, err)
require.EqualValues(t, 0, dr.DeleteCount)
remaining := queryRoaringIDs(t, ctx, mc, collectionName, "id >= 0", nil)
require.Len(t, remaining, membershipTotalRows, "empty positive delete must preserve every row")
})
// Negated set: delete everything outside the set.
t.Run("negated", func(t *testing.T) {
collectionName := common.GenRandomString("roaring_del_neg", 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRows(t, ctx, mc, collectionName, false)
flushLoadMembership(t, ctx, mc, collectionName)
keep := []int64{0, 1, 2}
blob, err := client.NewRoaringBitmapBlob(keep)
require.NoError(t, err)
dr, err := mc.Delete(ctx, client.NewDeleteOption(collectionName).
WithExpr(fmt.Sprintf("not membership_match(%s, {rb}, type=roaring)", membershipCreatorField)).
WithTemplateParam("rb", blob))
require.NoError(t, err)
// Negating the set deletes every creator value except len(keep), i.e.
// (domain - len(keep)) values; each value spans totalRows/domain rows.
require.EqualValues(t, int64((membershipDomain-len(keep))*membershipTotalRows/membershipDomain), dr.DeleteCount)
remaining := queryRoaringIDs(t, ctx, mc, collectionName, "id >= 0", nil)
expected := expectedMembershipIDs(membershipTotalRows, membershipDomain, map[int64]struct{}{
0: {}, 1: {}, 2: {},
})
require.ElementsMatch(t, expected, remaining,
"negated delete must preserve exactly the rows whose creator is in the keep set")
})
// Negated empty: every non-NULL row is outside the empty set, so delete all.
t.Run("negated_empty", func(t *testing.T) {
collectionName := common.GenRandomString("roaring_del_neg_empty", 6)
createIntMembershipCollection(t, ctx, mc, collectionName, false)
insertIntMembershipRows(t, ctx, mc, collectionName, false)
flushLoadMembership(t, ctx, mc, collectionName)
blob, err := client.NewRoaringBitmapBlob([]int64{})
require.NoError(t, err)
dr, err := mc.Delete(ctx, client.NewDeleteOption(collectionName).
WithExpr(fmt.Sprintf("not membership_match(%s, {rb}, type=roaring)", membershipCreatorField)).
WithTemplateParam("rb", blob))
require.NoError(t, err)
require.EqualValues(t, membershipTotalRows, dr.DeleteCount)
remaining := queryRoaringIDs(t, ctx, mc, collectionName, "id >= 0", nil)
require.Empty(t, remaining, "negated empty delete must remove every row")
})
}
@@ -20,7 +20,6 @@
package testcases
import (
"context"
"fmt"
"math/rand"
"strings"
@@ -33,7 +32,6 @@ import (
"github.com/milvus-io/milvus/client/v3/entity"
"github.com/milvus-io/milvus/client/v3/index"
client "github.com/milvus-io/milvus/client/v3/milvusclient"
"github.com/milvus-io/milvus/tests/go_client/base"
"github.com/milvus-io/milvus/tests/go_client/common"
hp "github.com/milvus-io/milvus/tests/go_client/testcases/helper"
)
@@ -87,12 +85,6 @@ func indexAndLoad(t *testing.T, ctx CtxT, mc MC, collName string, dim int) {
common.CheckErr(t, loadTask.Await(ctx), true)
}
// type aliases to keep test signatures readable
type (
CtxT = context.Context
MC = *base.MilvusClient
)
// TestStructArrayCreateWithClipEmbedding1 ports test_create_struct_array_with_clip_embedding1.
func TestStructArrayCreateWithClipEmbedding1(t *testing.T) {
ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
@@ -30,6 +30,7 @@ import (
"github.com/milvus-io/milvus-proto/go-api/v3/schemapb"
"github.com/milvus-io/milvus/client/v3/milvusclient"
"github.com/milvus-io/milvus/pkg/v3/common"
"github.com/milvus-io/milvus/pkg/v3/proto/indexpb"
"github.com/milvus-io/milvus/pkg/v3/util/funcutil"
"github.com/milvus-io/milvus/pkg/v3/util/merr"
"github.com/milvus-io/milvus/pkg/v3/util/metric"
@@ -58,6 +59,11 @@ const (
creatorID32Field = "creatorId32" // INT32
creatorIDStrField = "creatorIdStr"
metaJSONField = "meta"
// HYBRID is an internal-only scalar index type. Exercise it through the
// supported AUTOINDEX entry point and pin the resolution policy used by
// this suite so the tested physical path cannot drift with defaults.
scalarAutoIndexBuildParams = `{"int":"HYBRID","varchar":"HYBRID","bool":"BITMAP","float":"HYBRID","json":"HYBRID","geometry":"RTREE","timestamptz":"STL_SORT"}`
)
// intWidthFields are the integer fields (each carrying the same creatorId value,
@@ -71,6 +77,7 @@ func (s *BloomMatchTestSuite) SetupSuite() {
// it (default is off). Harmless for the other tests, whose fields keep their
// raw data. Must be set before the cluster starts.
s.WithMilvusConfig(paramtable.Get().QueryNodeCfg.IndexOffsetCacheEnabled.Key, "true")
s.WithMilvusConfig(paramtable.Get().AutoIndexConfig.ScalarAutoIndexParams.Key, scalarAutoIndexBuildParams)
s.MiniClusterSuite.SetupSuite()
s.dbName = ""
s.dim = 128
@@ -342,9 +349,40 @@ func (s *BloomMatchTestSuite) insertGrowing(collectionName string, creators []in
// scalarIndexSpec is one scalar index to build on a field.
type scalarIndexSpec struct {
field string
indexName string
params []*commonpb.KeyValuePair
field string
indexName string
params []*commonpb.KeyValuePair
expectedBuiltType string
}
// assertBuiltIndexTypes checks DataCoord's build parameters rather than the
// user-facing DescribeIndex response, which deliberately preserves AUTOINDEX.
// This makes the internal HYBRID coverage explicit without exposing HYBRID as
// a user-creatable index type.
func (s *BloomMatchTestSuite) assertBuiltIndexTypes(ctx context.Context, collectionName string, specs []scalarIndexSpec) {
collectionResp, err := s.Cluster.MilvusClient.DescribeCollection(ctx, &milvuspb.DescribeCollectionRequest{
DbName: s.dbName,
CollectionName: collectionName,
})
s.Require().NoError(err)
s.Require().NoError(merr.Error(collectionResp.GetStatus()))
for _, spec := range specs {
if spec.expectedBuiltType == "" {
continue
}
indexResp, err := s.Cluster.MixCoordClient.DescribeIndex(ctx, &indexpb.DescribeIndexRequest{
CollectionID: collectionResp.GetCollectionID(),
IndexName: spec.indexName,
})
s.Require().NoError(err)
s.Require().NoError(merr.Error(indexResp.GetStatus()))
s.Require().Lenf(indexResp.GetIndexInfos(), 1, "describe internal index %s", spec.indexName)
builtType, err := funcutil.GetAttrByKeyFromRepeatedKV(common.IndexTypeKey, indexResp.GetIndexInfos()[0].GetIndexParams())
s.Require().NoError(err)
s.Equalf(spec.expectedBuiltType, builtType, "built index type for %s", spec.indexName)
}
}
// buildScalarIndexesAndReload releases the already-loaded collection, builds every
@@ -378,6 +416,7 @@ func (s *BloomMatchTestSuite) buildScalarIndexesAndReload(collectionName string,
for _, spec := range specs {
s.WaitForIndexBuiltWithIndexName(ctx, collectionName, spec.field, spec.indexName)
}
s.assertBuiltIndexTypes(ctx, collectionName, specs)
loadResp, err := c.MilvusClient.LoadCollection(ctx, &milvuspb.LoadCollectionRequest{
DbName: s.dbName,
@@ -690,39 +729,47 @@ func (s *BloomMatchTestSuite) TestQueryScalarIndexTypeMatrix() {
strBlob := s.bloomBlobStr(strMembers)
rounds := []struct {
indexType string
onInts bool
onVarchar bool
name string
requestIndexType string
expectedBuiltType string
onInts bool
onVarchar bool
}{
{"STL_SORT", true, true},
{"BITMAP", true, true},
{"INVERTED", true, true},
{"HYBRID", true, true},
{"TRIE", false, true}, // varchar only
{"STL_SORT", "STL_SORT", "STL_SORT", true, true},
{"BITMAP", "BITMAP", "BITMAP", true, true},
{"INVERTED", "INVERTED", "INVERTED", true, true},
{"HYBRID", "AUTOINDEX", "HYBRID", true, true},
{"TRIE", "TRIE", "TRIE", false, true}, // varchar only
}
for _, r := range rounds {
s.Run(r.indexType, func() {
collectionName := "test_bloom_match_idx_" + strings.ToLower(r.indexType) + "_" + funcutil.GenRandomStr()
s.Run(r.name, func() {
collectionName := "test_bloom_match_idx_" + strings.ToLower(r.name) + "_" + funcutil.GenRandomStr()
s.setupCollection(collectionName)
var specs []scalarIndexSpec
if r.onInts {
for _, f := range intWidthFields {
specs = append(specs, scalarIndexSpec{field: f, indexName: "idx_" + f, params: scalarIndexParams(r.indexType)})
specs = append(specs, scalarIndexSpec{
field: f, indexName: "idx_" + f,
params: scalarIndexParams(r.requestIndexType), expectedBuiltType: r.expectedBuiltType,
})
}
}
if r.onVarchar {
specs = append(specs, scalarIndexSpec{field: creatorIDStrField, indexName: "idx_" + creatorIDStrField, params: scalarIndexParams(r.indexType)})
specs = append(specs, scalarIndexSpec{
field: creatorIDStrField, indexName: "idx_" + creatorIDStrField,
params: scalarIndexParams(r.requestIndexType), expectedBuiltType: r.expectedBuiltType,
})
}
s.buildScalarIndexesAndReload(collectionName, specs)
if r.onInts {
for _, f := range intWidthFields {
s.assertBloomSupersetInt(collectionName, f, intMembers, intBlob, r.indexType+"/"+f)
s.assertBloomSupersetInt(collectionName, f, intMembers, intBlob, r.name+"/"+f)
}
}
if r.onVarchar {
s.assertBloomSupersetStr(collectionName, creatorIDStrField, strMembers, strBlob, r.indexType+"/varchar")
s.assertBloomSupersetStr(collectionName, creatorIDStrField, strMembers, strBlob, r.name+"/varchar")
}
})
}
@@ -735,20 +782,29 @@ func (s *BloomMatchTestSuite) TestQueryScalarIndexTypeMatrix() {
// yet the raw JSON survives — the direct refutation of the "index drops raw JSON,
// bloom fails after seal" concern.
func (s *BloomMatchTestSuite) TestQueryJsonPathIndexTypeMatrix() {
for _, indexType := range []string{"INVERTED", "HYBRID"} {
s.Run(indexType, func() {
collectionName := "test_bloom_match_jsonidx_" + strings.ToLower(indexType) + "_" + funcutil.GenRandomStr()
rounds := []struct {
name string
requestIndexType string
expectedBuiltType string
}{
{"INVERTED", "INVERTED", "INVERTED"},
{"HYBRID", "AUTOINDEX", "HYBRID"},
}
for _, r := range rounds {
s.Run(r.name, func() {
collectionName := "test_bloom_match_jsonidx_" + strings.ToLower(r.name) + "_" + funcutil.GenRandomStr()
creators, _ := s.setupCollection(collectionName)
s.buildScalarIndexesAndReload(collectionName, []scalarIndexSpec{{
field: metaJSONField,
indexName: "idx_meta_creator",
field: metaJSONField,
indexName: "idx_meta_creator",
expectedBuiltType: r.expectedBuiltType,
params: []*commonpb.KeyValuePair{
{Key: common.IndexTypeKey, Value: indexType},
{Key: common.IndexTypeKey, Value: r.requestIndexType},
{Key: "json_path", Value: fmt.Sprintf(`%s["creator"]`, metaJSONField)},
{Key: "json_cast_type", Value: "double"},
},
}})
s.assertJSONPathStrictMembership(collectionName, creators, indexType)
s.assertJSONPathStrictMembership(collectionName, creators, r.name)
})
}
}
File diff suppressed because it is too large. Load diff
+2 -2
View File
@@ -26,8 +26,8 @@ pytest-sugar==0.9.5
pytest-random-order
# pymilvus
pymilvus==3.1.0rc80
pymilvus[bulk_writer]==3.1.0rc80
pymilvus==3.1.0rc83
pymilvus[bulk_writer]==3.1.0rc83
# for protobuf
protobuf>=5.29.5
@@ -1,17 +1,10 @@
from pymilvus import DataType
success = "success"
class FMINDEX:
# FM-index is an exact byte-level substring index for VARCHAR. JSON string
# paths, TEXT and ARRAY are follow-ups (rejected at create_index for now).
supported_field_types = [
DataType.VARCHAR,
]
# Build-parameter test configurations. fm_sa_sample_rate is optional
# (defaults to 8) and, when present, must be an integer in [4, 256].
# Build-parameter test configurations. fm_sa_sample_rate defaults to 8 and
# must be an integer in [4, 256]; fm_block_bytes defaults to 64 and must be
# an integer power of two in [8, 128].
build_params = [
{
"description": "no optional params (defaults applied)",
@@ -65,6 +58,16 @@ class FMINDEX:
"params": {"fm_block_bytes": 128},
"expected": success,
},
{
"description": "sample rate and block bytes supplied together",
"params": {"fm_sa_sample_rate": 32, "fm_block_bytes": 16},
"expected": success,
},
{
"description": "block bytes below minimum (power of two but < 8)",
"params": {"fm_block_bytes": 4},
"expected": {"err_code": 1100, "err_msg": "fm_block_bytes for FM-index must be a power of two"},
},
{
"description": "block bytes not a power of two",
"params": {"fm_block_bytes": 24},
@@ -75,4 +78,9 @@ class FMINDEX:
"params": {"fm_block_bytes": 256},
"expected": {"err_code": 1100, "err_msg": "fm_block_bytes for FM-index must be a power of two"},
},
{
"description": "block bytes not an integer",
"params": {"fm_block_bytes": "abc"},
"expected": {"err_code": 1100, "err_msg": "fm_block_bytes for FM-index must be an integer"},
},
]
@@ -23,8 +23,9 @@ class TestFMIndexBuildParams(TestMilvusClientV2Base):
@pytest.mark.parametrize("params", FMINDEX.build_params)
def test_fmindex_build_params(self, params):
"""
Build FMINDEX with a matrix of fm_sa_sample_rate values; valid ones
succeed and are persisted, invalid ones are rejected at create_index.
Build FMINDEX with a matrix of fm_sa_sample_rate and fm_block_bytes
values; valid ones succeed and are persisted, invalid ones are rejected
at create_index.
"""
client = self._client()
collection_name = cf.gen_collection_name_by_testcase_name()
@@ -32,14 +33,7 @@ class TestFMIndexBuildParams(TestMilvusClientV2Base):
schema.add_field(pk_field_name, datatype=DataType.INT64, is_primary=True, auto_id=False)
schema.add_field(vector_field_name, datatype=DataType.FLOAT_VECTOR, dim=dim)
schema.add_field(content_field_name, datatype=DataType.VARCHAR, max_length=100)
self.create_collection(client, collection_name, schema=schema)
nb = default_nb
rows = cf.gen_row_data_by_schema(nb=nb, schema=schema, start=0)
for i, row in enumerate(rows):
row[content_field_name] = f"The {content_keywords[i % len(content_keywords)]} number {i}"
self.insert(client, collection_name, rows)
self.flush(client, collection_name)
self.create_collection(client, collection_name, schema=schema, consistency_level="Strong")
build_params = params.get("params", None)
index_params = self.prepare_index_params(client)[0]
@@ -49,45 +43,65 @@ class TestFMIndexBuildParams(TestMilvusClientV2Base):
)
if params.get("expected", None) != success:
self.create_index(
error, _ = self.create_index(
client, collection_name, index_params, check_task=CheckTasks.err_res, check_items=params.get("expected")
)
assert error.code == params["expected"][ct.err_code], error
return
self.create_index(client, collection_name, index_params)
self.wait_for_index_ready(client, collection_name, index_name=index_name)
nb = default_nb
rows = cf.gen_row_data_by_schema(nb=nb, schema=schema, start=0)
for i, row in enumerate(rows):
row[content_field_name] = f"The {content_keywords[i % len(content_keywords)]} number {i}"
self.insert(client, collection_name, rows)
self.flush(client, collection_name)
# persisted params (only fm_sa_sample_rate when explicitly set)
self.create_index(client, collection_name, index_params)
assert self.wait_for_index_ready(client, collection_name, index_name=index_name)
# Persisted parameters must retain their own key/value association.
idx_info = client.describe_index(collection_name, index_name)
assert idx_info["index_type"] == index_type
assert idx_info["pending_index_rows"] == 0
assert idx_info["indexed_rows"] == nb
if build_params:
for key, value in build_params.items():
assert key in idx_info.keys()
assert str(value) in idx_info.values()
assert idx_info[key] == str(value)
@pytest.mark.tags(CaseLabel.L1)
def test_fmindex_on_non_varchar_field_rejected(self):
@pytest.mark.parametrize(
"field_name, datatype",
[
("int_field", DataType.INT64),
("bool_field", DataType.BOOL),
("float_field", DataType.FLOAT),
("double_field", DataType.DOUBLE),
],
)
def test_fmindex_on_non_varchar_field_rejected(self, field_name, datatype):
"""
FMINDEX is VARCHAR-only in this release; building it on an INT64 field
(or any non-VARCHAR field such as JSON) must be rejected.
FMINDEX is VARCHAR-only in this release; building it on a numeric / bool
field (INT64/BOOL/FLOAT/DOUBLE, and JSON separately below) must be
rejected.
"""
client = self._client()
collection_name = cf.gen_collection_name_by_testcase_name()
schema, _ = self.create_schema(client)
schema.add_field(pk_field_name, datatype=DataType.INT64, is_primary=True, auto_id=False)
schema.add_field(vector_field_name, datatype=DataType.FLOAT_VECTOR, dim=dim)
schema.add_field("int_field", datatype=DataType.INT64)
self.create_collection(client, collection_name, schema=schema)
schema.add_field(field_name, datatype=datatype)
self.create_collection(client, collection_name, schema=schema, consistency_level="Strong")
index_params = self.prepare_index_params(client)[0]
index_params.add_index(field_name="int_field", index_name="fm_bad", index_type=index_type, params={})
self.create_index(
index_params.add_index(field_name=field_name, index_name="fm_bad", index_type=index_type, params={})
error, _ = self.create_index(
client,
collection_name,
index_params,
check_task=CheckTasks.err_res,
check_items={"err_code": 1100, "err_msg": "FM-index can only be created on VARCHAR field"},
)
assert error.code == 1100, error
@pytest.mark.tags(CaseLabel.L1)
def test_fmindex_on_json_field_rejected(self):
@@ -101,7 +115,7 @@ class TestFMIndexBuildParams(TestMilvusClientV2Base):
schema.add_field(pk_field_name, datatype=DataType.INT64, is_primary=True, auto_id=False)
schema.add_field(vector_field_name, datatype=DataType.FLOAT_VECTOR, dim=dim)
schema.add_field("json_field", datatype=DataType.JSON)
self.create_collection(client, collection_name, schema=schema)
self.create_collection(client, collection_name, schema=schema, consistency_level="Strong")
index_params = self.prepare_index_params(client)[0]
index_params.add_index(
@@ -118,26 +132,77 @@ class TestFMIndexBuildParams(TestMilvusClientV2Base):
check_items={"err_code": 1100, "err_msg": "FM-index can only be created on VARCHAR field"},
)
class TestFMIndexQuery(TestMilvusClientV2Base):
def _build_loaded_collection(self, client):
"""Create a collection with an FMINDEX field and an identical un-indexed
field, insert keyword data, flush (sealed), build indexes and load."""
@pytest.mark.tags(CaseLabel.L1)
def test_fmindex_on_struct_sub_field_rejected(self):
"""
A struct-array sub-field (e.g. structA[str_val] / structA[int_val]) is an
ARRAY field to the checker regardless of the element type; FMINDEX is
VARCHAR-only, so building on it must be rejected.
"""
client = self._client()
collection_name = cf.gen_collection_name_by_testcase_name()
schema, _ = self.create_schema(client)
schema.add_field(pk_field_name, datatype=DataType.INT64, is_primary=True, auto_id=False)
schema.add_field(vector_field_name, datatype=DataType.FLOAT_VECTOR, dim=dim)
schema.add_field(no_index_field_name, datatype=DataType.VARCHAR, max_length=20)
schema.add_field(content_field_name, datatype=DataType.VARCHAR, max_length=20)
self.create_collection(client, collection_name, schema=schema)
struct_schema = self.create_struct_field_schema(client)[0]
struct_schema.add_field("str_val", DataType.VARCHAR, max_length=64)
struct_schema.add_field("int_val", DataType.INT64)
schema.add_field(
"structA",
datatype=DataType.ARRAY,
element_type=DataType.STRUCT,
struct_schema=struct_schema,
max_capacity=10,
)
self.create_collection(client, collection_name, schema=schema, consistency_level="Strong")
for field in ["structA[str_val]", "structA[int_val]"]:
index_params = self.prepare_index_params(client)[0]
index_params.add_index(field_name=field, index_name="fm_struct", index_type=index_type, params={})
error, _ = self.create_index(
client,
collection_name,
index_params,
check_task=CheckTasks.err_res,
check_items={"err_code": 1100, "err_msg": "FM-index can only be created on VARCHAR field"},
)
assert error.code == 1100, error
class TestFMIndexQuery(TestMilvusClientV2Base):
def _assert_index_ready(self, client, collection_name, index_name, expected_rows, expected_index_type):
assert self.wait_for_index_ready(client, collection_name, index_name=index_name), (
f"index {index_name} on {collection_name} did not become ready"
)
info = self.describe_index(client, collection_name, index_name)[0]
assert info["index_name"] == index_name, info
assert info["field_name"] == index_name, info
assert info["index_type"] == expected_index_type, info
assert info["state"] == "Finished", info
assert info["pending_index_rows"] == 0, info
assert info["total_rows"] == expected_rows, info
assert info["indexed_rows"] == expected_rows, info
def _build_loaded_collection(self, client):
"""Create a collection with an FMINDEX field and an identical un-indexed
field, insert keyword data, flush (sealed), build indexes and load."""
collection_name = cf.gen_collection_name_by_testcase_name(module_index=2)
schema, _ = self.create_schema(client)
schema.add_field(pk_field_name, datatype=DataType.INT64, is_primary=True, auto_id=False)
schema.add_field(vector_field_name, datatype=DataType.FLOAT_VECTOR, dim=dim)
schema.add_field(no_index_field_name, datatype=DataType.VARCHAR, max_length=600)
schema.add_field(content_field_name, datatype=DataType.VARCHAR, max_length=600)
self.create_collection(client, collection_name, schema=schema, consistency_level="Strong")
insert_times = 2
filler = "y" * 500
for t in range(insert_times):
rows = cf.gen_row_data_by_schema(nb=default_nb, schema=schema, start=t * default_nb)
for j, row in enumerate(rows):
kw = content_keywords[j % len(content_keywords)]
row[no_index_field_name] = kw
row[content_field_name] = kw
text = "stadium" if j % 500 == 0 else filler
row[no_index_field_name] = text
row[content_field_name] = text
self.insert(client, collection_name, rows)
self.flush(client, collection_name)
@@ -147,22 +212,79 @@ class TestFMIndexQuery(TestMilvusClientV2Base):
)
index_params.add_index(field_name=content_field_name, index_type=index_type, params={"fm_sa_sample_rate": 32})
self.create_index(client, collection_name, index_params)
self.wait_for_index_ready(client, collection_name, index_name=vector_field_name)
self.wait_for_index_ready(client, collection_name, index_name=content_field_name)
total_rows = insert_times * default_nb
self._assert_index_ready(client, collection_name, vector_field_name, total_rows, "IVF_FLAT")
self._assert_index_ready(client, collection_name, content_field_name, total_rows, "FMINDEX")
self.load_collection(client, collection_name)
return collection_name, insert_times, schema
def _assert_same(self, client, collection_name, indexed_expr, scan_expr, **kwargs):
"""The FMINDEX-accelerated query must return exactly the same rows as the
"""The indexed-field query must return exactly the same rows as the
brute-force scan over the un-indexed twin field."""
res_idx = self.query(client, collection_name, filter=indexed_expr, output_fields=["id"], **kwargs)[0]
res_scan = self.query(client, collection_name, filter=scan_expr, output_fields=["id"], **kwargs)[0]
ids_idx = sorted(r["id"] for r in res_idx)
ids_scan = sorted(r["id"] for r in res_scan)
res_idx = self.query(client, collection_name, filter=indexed_expr, output_fields=[pk_field_name], **kwargs)[0]
res_scan = self.query(client, collection_name, filter=scan_expr, output_fields=[pk_field_name], **kwargs)[0]
ids_idx = sorted(r[pk_field_name] for r in res_idx)
ids_scan = sorted(r[pk_field_name] for r in res_scan)
assert ids_idx == ids_scan
return ids_idx
@pytest.mark.tags(CaseLabel.L1)
def _search_ids(self, client, collection_name, filter_expr, limit, expected_content):
"""Run a complete-candidate Search so its scalar filter can be compared
with Query on the same loaded collection and indexes."""
result_sets = client.search(
collection_name,
data=[[1.0] * dim],
anns_field=vector_field_name,
search_params={"metric_type": "COSINE", "params": {"nprobe": 128}},
limit=limit,
filter=filter_expr,
output_fields=[no_index_field_name],
)
assert len(result_sets) == 1
hits = result_sets[0]
ids = [hit[pk_field_name] for hit in hits]
assert len(ids) == len(set(ids)), "Search returned duplicate primary keys"
distances = [hit["distance"] for hit in hits]
assert distances == sorted(distances, reverse=True), "COSINE Search scores must be nonincreasing"
for hit in hits:
entity = hit.get("entity")
assert isinstance(entity, dict), hit
assert set(entity) == {no_index_field_name}, hit
assert entity[no_index_field_name] == expected_content, hit
return sorted(ids)
def _build_twin_collection(self, client, content_fn, max_length=64, total_nb=default_nb):
"""Create a collection with an FMINDEX field and an identical un-indexed
twin field, filled by content_fn(i) -> str, flush (sealed), build indexes
and load. Returns the collection name."""
collection_name = cf.gen_collection_name_by_testcase_name(module_index=2)
schema, _ = self.create_schema(client)
schema.add_field(pk_field_name, datatype=DataType.INT64, is_primary=True, auto_id=False)
schema.add_field(vector_field_name, datatype=DataType.FLOAT_VECTOR, dim=dim)
schema.add_field(no_index_field_name, datatype=DataType.VARCHAR, max_length=max_length)
schema.add_field(content_field_name, datatype=DataType.VARCHAR, max_length=max_length)
self.create_collection(client, collection_name, schema=schema, consistency_level="Strong")
rows = cf.gen_row_data_by_schema(nb=total_nb, schema=schema, start=0)
for i, row in enumerate(rows):
text = content_fn(i)
row[no_index_field_name] = text
row[content_field_name] = text
self.insert(client, collection_name, rows)
self.flush(client, collection_name)
index_params = self.prepare_index_params(client)[0]
index_params.add_index(
field_name=vector_field_name, metric_type="COSINE", index_type="IVF_FLAT", params={"nlist": 128}
)
index_params.add_index(field_name=content_field_name, index_type=index_type, params={"fm_sa_sample_rate": 32})
self.create_index(client, collection_name, index_params)
self._assert_index_ready(client, collection_name, vector_field_name, total_nb, "IVF_FLAT")
self._assert_index_ready(client, collection_name, content_field_name, total_nb, "FMINDEX")
self.load_collection(client, collection_name)
return collection_name
@pytest.mark.tags(CaseLabel.L0)
def test_fmindex_prefix_infix_suffix(self):
"""
Exact prefix / infix / suffix LIKE on the FMINDEX field must match the
@@ -170,23 +292,25 @@ class TestFMIndexQuery(TestMilvusClientV2Base):
"""
client = self._client()
collection_name, insert_times, _ = self._build_loaded_collection(client)
expected = insert_times * default_nb // len(content_keywords) # rows per keyword
expected_ids = sorted(
t * default_nb + j for t in range(insert_times) for j in range(default_nb) if j % 500 == 0
)
# prefix: LIKE 'sta%'
ids = self._assert_same(
client, collection_name, f'{content_field_name} LIKE "sta%"', f'{no_index_field_name} LIKE "sta%"'
)
assert len(ids) == expected
assert ids == expected_ids
# suffix: LIKE '%ium'
ids = self._assert_same(
client, collection_name, f'{content_field_name} LIKE "%ium"', f'{no_index_field_name} LIKE "%ium"'
)
assert len(ids) == expected
assert ids == expected_ids
# infix: LIKE '%adi%'
ids = self._assert_same(
client, collection_name, f'{content_field_name} LIKE "%adi%"', f'{no_index_field_name} LIKE "%adi%"'
)
assert len(ids) == expected
assert ids == expected_ids
# no match
ids = self._assert_same(
client, collection_name, f'{content_field_name} LIKE "zzz%"', f'{no_index_field_name} LIKE "zzz%"'
@@ -195,9 +319,24 @@ class TestFMIndexQuery(TestMilvusClientV2Base):
# exact equality is NOT accelerated by FMINDEX (it declines ==/IN and
# falls back to the raw-data scan) but must still return correct rows
ids = self._assert_same(
client, collection_name, f'{content_field_name} == "park"', f'{no_index_field_name} == "park"'
client, collection_name, f'{content_field_name} == "stadium"', f'{no_index_field_name} == "stadium"'
)
assert len(ids) == expected
assert ids == expected_ids
# Search and Query intentionally reuse this one loaded collection. With
# IVF nprobe covering every list and limit covering all rows, the
# FMINDEX filter result must match the raw-scan twin exactly.
search_ids = self._search_ids(
client,
collection_name,
f'{content_field_name} LIKE "sta%"',
limit=insert_times * default_nb,
expected_content="stadium",
)
scan_ids = self._assert_same(
client, collection_name, f'{content_field_name} LIKE "sta%"', f'{no_index_field_name} LIKE "sta%"'
)
assert search_ids == scan_ids
@pytest.mark.tags(CaseLabel.L1)
def test_fmindex_growing_and_sealed_mixed(self):
@@ -213,14 +352,14 @@ class TestFMIndexQuery(TestMilvusClientV2Base):
# extra batch that stays in a GROWING segment (no flush after it)
start = insert_times * default_nb
rows = cf.gen_row_data_by_schema(nb=default_nb, schema=schema, start=start)
filler = "y" * 500
for j, row in enumerate(rows):
kw = content_keywords[j % len(content_keywords)]
row[no_index_field_name] = kw
row[content_field_name] = kw
text = "stadium" if j % 500 == 0 else filler
row[no_index_field_name] = text
row[content_field_name] = text
self.insert(client, collection_name, rows)
# Strong consistency so the un-flushed growing rows are visible.
expected = (insert_times + 1) * default_nb // len(content_keywords)
ids = self._assert_same(
client,
collection_name,
@@ -228,20 +367,20 @@ class TestFMIndexQuery(TestMilvusClientV2Base):
f'{no_index_field_name} LIKE "sta%"',
consistency_level="Strong",
)
assert len(ids) == expected
expected = sorted(
batch * default_nb + j for batch in range(insert_times + 1) for j in range(default_nb) if j % 500 == 0
)
assert ids == expected
@pytest.mark.tags(CaseLabel.L1)
def test_fmindex_accelerated_path_long_text_low_hit(self):
def test_fmindex_long_text_low_hit_differential_correctness(self):
"""
Positive accelerated-path case. The other query tests use short keyword
values where every keyword matches ~1/8 of the rows, so the count-first
cost guard declines and they fall back to the scan (still exact, but they
never exercise FMINDEX's own execution path). Here the corpus is long text
(~500 chars/row) with a rare marker in only a handful of rows: total
tokens are large and the marker's occurrence count is tiny, so the guard
ACCEPTS the pattern and the query is actually answered by FMINDEX. The
result must still equal the brute-force scan on the twin field and be
non-empty.
Differential correctness case for a low-hit, long-text corpus. Each row
is ~500 characters and the marker occurs only once per 500 rows, so the
token population is large while the candidate count stays small. This
makes the count-first cost guard eligible to accept the pattern. The
public test API has no stable execution-path signal, so this test proves
result correctness against the twin-field scan only.
"""
client = self._client()
collection_name = cf.gen_collection_name_by_testcase_name()
@@ -250,7 +389,7 @@ class TestFMIndexQuery(TestMilvusClientV2Base):
schema.add_field(vector_field_name, datatype=DataType.FLOAT_VECTOR, dim=dim)
schema.add_field(no_index_field_name, datatype=DataType.VARCHAR, max_length=600)
schema.add_field(content_field_name, datatype=DataType.VARCHAR, max_length=600)
self.create_collection(client, collection_name, schema=schema)
self.create_collection(client, collection_name, schema=schema, consistency_level="Strong")
nb = default_nb
filler = "y" * 500 # marker never occurs in the filler
@@ -272,12 +411,12 @@ class TestFMIndexQuery(TestMilvusClientV2Base):
)
index_params.add_index(field_name=content_field_name, index_type=index_type, params={"fm_sa_sample_rate": 32})
self.create_index(client, collection_name, index_params)
self.wait_for_index_ready(client, collection_name, index_name=vector_field_name)
self.wait_for_index_ready(client, collection_name, index_name=content_field_name)
self._assert_index_ready(client, collection_name, vector_field_name, nb, "IVF_FLAT")
self._assert_index_ready(client, collection_name, content_field_name, nb, "FMINDEX")
self.load_collection(client, collection_name)
# low-hit infix over long text: the guard accepts -> FMINDEX path, and the
# result must match the brute-force scan on the twin field (and be non-empty)
# Low-hit infix over long text: compare the indexed field with the
# un-indexed twin. Physical FMINDEX execution is not asserted here.
ids = self._assert_same(
client,
collection_name,
@@ -302,9 +441,9 @@ class TestFMIndexQuery(TestMilvusClientV2Base):
schema.add_field(vector_field_name, datatype=DataType.FLOAT_VECTOR, dim=dim)
schema.add_field(no_index_field_name, datatype=DataType.VARCHAR, max_length=600, nullable=True)
schema.add_field(content_field_name, datatype=DataType.VARCHAR, max_length=600, nullable=True)
self.create_collection(client, collection_name, schema=schema)
self.create_collection(client, collection_name, schema=schema, consistency_level="Strong")
batch_nb = 1000
batch_nb = 1500
filler = "y" * 500
matching_ids = set()
qop_only_ids = set()
@@ -337,8 +476,8 @@ class TestFMIndexQuery(TestMilvusClientV2Base):
self.insert(client, collection_name, rows)
self.flush(client, collection_name)
# Each fragment occurs only eight times in roughly one million indexed
# tokens: 8 * sample_rate(32) is below the default 0.001 cost threshold,
# Each fragment occurs 12 times in roughly 1.5 million indexed tokens:
# 12 * sample_rate(32) = 384 is below the default 0.001 cost threshold (~1500),
# so this interior-wildcard expression takes the FMINDEX Match path.
index_params = self.prepare_index_params(client)[0]
index_params.add_index(
@@ -346,8 +485,9 @@ class TestFMIndexQuery(TestMilvusClientV2Base):
)
index_params.add_index(field_name=content_field_name, index_type=index_type, params={"fm_sa_sample_rate": 32})
self.create_index(client, collection_name, index_params)
self.wait_for_index_ready(client, collection_name, index_name=vector_field_name)
self.wait_for_index_ready(client, collection_name, index_name=content_field_name)
total_rows = 2 * batch_nb
self._assert_index_ready(client, collection_name, vector_field_name, total_rows, "IVF_FLAT")
self._assert_index_ready(client, collection_name, content_field_name, total_rows, "FMINDEX")
self.load_collection(client, collection_name)
ids = self._assert_same(
@@ -367,3 +507,160 @@ class TestFMIndexQuery(TestMilvusClientV2Base):
f'{no_index_field_name} LIKE ""',
)
assert set(ids) == empty_ids
@pytest.mark.tags(CaseLabel.L1)
def test_fmindex_null_rows_not_matched(self):
"""
Nullable VARCHAR NULL rows are treated as an empty document: no pattern,
not even LIKE '%', may match them.
"""
client = self._client()
collection_name = cf.gen_collection_name_by_testcase_name()
schema, _ = self.create_schema(client)
schema.add_field(pk_field_name, datatype=DataType.INT64, is_primary=True, auto_id=False)
schema.add_field(vector_field_name, datatype=DataType.FLOAT_VECTOR, dim=dim)
schema.add_field(content_field_name, datatype=DataType.VARCHAR, max_length=64, nullable=True)
self.create_collection(client, collection_name, schema=schema, consistency_level="Strong")
total_nb = default_nb
rows = cf.gen_row_data_by_schema(nb=total_nb, schema=schema, start=0)
non_null = 0
for i, row in enumerate(rows):
if i % 8 == 7:
row[content_field_name] = None
else:
row[content_field_name] = content_keywords[i % len(content_keywords)]
non_null += 1
self.insert(client, collection_name, rows)
self.flush(client, collection_name)
index_params = self.prepare_index_params(client)[0]
index_params.add_index(
field_name=vector_field_name, metric_type="COSINE", index_type="IVF_FLAT", params={"nlist": 128}
)
index_params.add_index(field_name=content_field_name, index_type=index_type, params={"fm_sa_sample_rate": 32})
self.create_index(client, collection_name, index_params)
self._assert_index_ready(client, collection_name, vector_field_name, total_nb, "IVF_FLAT")
self._assert_index_ready(client, collection_name, content_field_name, total_nb, "FMINDEX")
self.load_collection(client, collection_name)
all_rows = self.query(
client, collection_name, filter=f'{content_field_name} LIKE "%"', output_fields=[pk_field_name]
)[0]
all_ids = sorted(row[pk_field_name] for row in all_rows)
expected_all = [i for i in range(total_nb) if i % 8 != 7]
assert len(expected_all) == non_null
assert all_ids == expected_all, "LIKE '%' must return the exact non-NULL PK set"
sta = self.query(
client, collection_name, filter=f'{content_field_name} LIKE "sta%"', output_fields=[pk_field_name]
)[0]
sta_ids = sorted(row[pk_field_name] for row in sta)
expected_sta = [i for i in range(total_nb) if i % 8 == 0]
assert sta_ids == expected_sta, "LIKE 'sta%' must return the exact non-NULL stadium PK set"
@pytest.mark.tags(CaseLabel.L1)
def test_fmindex_non_ascii(self):
"""
Byte-exact substring matching over multi-byte UTF-8 content (CJK / emoji):
a byte-level index must not mis-align on multi-byte sequences.
"""
client = self._client()
values = ["中文测试数据", "测试中文", "emoji😀测试", "纯英文english", "中文😀中文"]
filler = "y" * 500
def content_for_id(row_id):
value_index = row_id % 500
return values[value_index] if value_index < len(values) else filler
total_tokens = sum(len(content_for_id(i).encode("utf-8")) for i in range(default_nb))
collection_name = self._build_twin_collection(client, content_for_id, max_length=600)
cases = [
("%测试%", {0, 1, 2}),
("%😀%", {2, 4}),
("%中文😀%", {4}),
]
for pattern, matching_value_indexes in cases:
ids = self._assert_same(
client,
collection_name,
f'{content_field_name} LIKE "{pattern}"',
f'{no_index_field_name} LIKE "{pattern}"',
)
expected = [i for i in range(default_nb) if i % 500 in matching_value_indexes]
assert len(expected) * 32 < total_tokens / 1000, "fixture must satisfy the FMINDEX cost guard"
assert ids == expected, f"unexpected UTF-8 LIKE ground truth for {pattern}"
@pytest.mark.tags(CaseLabel.L1)
def test_fmindex_empty_pattern_and_general_fallback(self):
"""
`LIKE '%'` is optimized to an empty-pattern anchored operation and uses
FMINDEX's IsNotNull fast path. General LIKE with an interior wildcard, a
single-char `_` wildcard, and regex `=~` fall back to the scan. Every
form must stay exact against the twin field.
"""
client = self._client()
collection_name = self._build_twin_collection(client, lambda i: f"req-{i % 4}-error-{i % 10}")
all_rows = self.query(
client, collection_name, filter=f'{content_field_name} LIKE "%"', output_fields=[pk_field_name]
)[0]
expected = list(range(default_nb))
assert sorted(row[pk_field_name] for row in all_rows) == expected
ids = self._assert_same(
client,
collection_name,
f'{content_field_name} LIKE "req-%error%"',
f'{no_index_field_name} LIKE "req-%error%"',
)
assert ids == expected
ids = self._assert_same(
client,
collection_name,
f'{content_field_name} LIKE "req-_-error-_"',
f'{no_index_field_name} LIKE "req-_-error-_"',
)
assert ids == expected
ids = self._assert_same(
client,
collection_name,
f'{content_field_name} =~ "req-.-error-."',
f'{no_index_field_name} =~ "req-.-error-."',
)
assert ids == expected
@pytest.mark.tags(CaseLabel.L1)
def test_fmindex_escaped_wildcards(self):
r"""
LIKE escape handling: `\%` matches a literal '%', `\_` a literal '_', and
`\\` a literal backslash. Escaped literals are carried in the expression's
raw string (r"...") so the backslash reaches the LIKE layer verbatim.
"""
client = self._client()
values = ["100%done", "under_score", "back\\slash", "plain", "50%_mixed"]
filler = "y" * 500
def content_for_id(row_id):
value_index = row_id % 500
return values[value_index] if value_index < len(values) else filler
total_tokens = sum(len(content_for_id(i).encode("utf-8")) for i in range(default_nb))
collection_name = self._build_twin_collection(client, content_for_id, max_length=600)
cases = [
(r"%\%%", {0, 4}),
(r"%\_%", {1, 4}),
(r"%\\%", {2}),
]
for escaped, matching_value_indexes in cases:
ids = self._assert_same(
client,
collection_name,
rf'{content_field_name} LIKE r"{escaped}"',
rf'{no_index_field_name} LIKE r"{escaped}"',
)
expected = [i for i in range(default_nb) if i % 500 in matching_value_indexes]
assert len(expected) * 32 < total_tokens / 1000, "fixture must satisfy the FMINDEX cost guard"
assert ids == expected, f"unexpected escaped LIKE ground truth for {escaped}"
@@ -0,0 +1,297 @@
import json
import time
import pytest
from api.milvus import VectorClient
from base.testbase import TestBase
from utils.constant import CaseLabel
from utils.utils import gen_collection_name
FM_DIM = 4
FM_ROWS = 2048
class TestFMIndexAndMembershipFilters(TestBase):
"""RESTful v2 smoke coverage for FMINDEX and membership expressions.
The RESTful expression-template API accepts JSON-representable values but
currently has no JSON mapping to TemplateValue.BytesVal. membership_match
therefore has protocol-negative coverage here; positive blob coverage
belongs to Go and PyMilvus until REST exposes a bytes representation.
"""
def _create_collection(self, scenario, include_content=True):
name = f"{self.__class__.__name__}_{scenario}_{gen_collection_name()}"
fields = [
{"fieldName": "id", "dataType": "Int64", "isPrimary": True},
{"fieldName": "creator_id", "dataType": "Int64"},
{"fieldName": "vector", "dataType": "FloatVector", "elementTypeParams": {"dim": str(FM_DIM)}},
]
if include_content:
fields.insert(
1,
{
"fieldName": "content",
"dataType": "VarChar",
"elementTypeParams": {"max_length": "600"},
},
)
fields.insert(
2,
{
"fieldName": "content_no_index",
"dataType": "VarChar",
"elementTypeParams": {"max_length": "600"},
},
)
rsp = self.collection_client.collection_create(
{
"collectionName": name,
"schema": {"autoId": False, "enableDynamicField": False, "fields": fields},
}
)
assert rsp["code"] == 0, rsp
return name
@staticmethod
def _content_for_id(row_id):
filler = "y" * 500
case = row_id % 500
if case == 0:
return "stadium"
if case == 1:
return "x" * 250 + "school" + "x" * 250
if case == 2:
return filler[:-7] + "library"
return filler
def _insert_rows(self, name, with_content=True):
rows = []
for i in range(FM_ROWS):
row = {
"id": i,
"creator_id": i % 50,
"vector": [float(i), 0.0, 0.0, 0.0],
}
if with_content:
text = self._content_for_id(i)
row["content"] = text
row["content_no_index"] = text
rows.append(row)
for start in range(0, FM_ROWS, 512):
rsp = self.vector_client.vector_insert({"collectionName": name, "data": rows[start : start + 512]})
assert rsp["code"] == 0, rsp
assert rsp["data"]["insertCount"] == len(rows[start : start + 512]), rsp
rsp = self.collection_client.flush(name)
assert rsp["code"] == 0, rsp
def _index_and_load(self, name, add_fmindex=True):
index_params = [
{
"fieldName": "vector",
"indexName": "vector_index",
"indexType": "FLAT",
"metricType": "L2",
"params": {},
}
]
if add_fmindex:
index_params.append(
{
"fieldName": "content",
"indexName": "content_fmindex",
"indexType": "FMINDEX",
"params": {"fm_sa_sample_rate": 8, "fm_block_bytes": 64},
}
)
rsp = self.index_client.index_create({"collectionName": name, "indexParams": index_params})
assert rsp["code"] == 0, rsp
for params in index_params:
index = self._wait_index_ready(name, params["indexName"], timeout=180)
assert index["indexState"] == "Finished", index
assert index.get("failReason", "") == "", index
assert index["fieldName"] == params["fieldName"], index
assert index["indexType"] == params["indexType"], index
assert index["totalRows"] == FM_ROWS, index
assert index["indexedRows"] == FM_ROWS, index
assert index["pendingRows"] == 0, index
if params["indexType"] == "FMINDEX":
reported = {item["key"]: item["value"] for item in index["indexParams"]}
assert reported["index_type"] == "FMINDEX", index
assert json.loads(reported["params"]) == params["params"], index
rsp = self.collection_client.collection_load(collection_name=name)
assert rsp["code"] == 0, rsp
self.collection_client.wait_load_completed(name, timeout=180)
describe = self.collection_client.collection_describe(name)
assert describe["code"] == 0, describe
assert describe["data"]["load"] == "LoadStateLoaded", describe
def _wait_index_ready(self, name, index_name, timeout):
start = time.time()
while time.time() - start < timeout:
rsp = self.index_client.index_describe(collection_name=name, index_name=index_name)
assert rsp["code"] == 0, rsp
assert len(rsp["data"]) == 1, rsp
index = rsp["data"][0]
if index["indexState"] == "Finished":
return index
if index["indexState"] == "Failed":
raise AssertionError(index)
time.sleep(1)
raise AssertionError(f"index {index_name} of collection {name} not ready after {timeout}s")
def _query_ids(self, name, filter_expr):
rsp = self.vector_client.vector_query(
{"collectionName": name, "filter": filter_expr, "outputFields": ["id"], "limit": FM_ROWS},
timeout=180,
)
assert rsp["code"] == 0, rsp
return sorted(row["id"] for row in rsp.get("data", []))
def _search_ids(self, name, filter_expr):
rsp = self.vector_client.vector_search(
{
"collectionName": name,
"data": [[1.0] * FM_DIM],
"annsField": "vector",
"filter": filter_expr,
"outputFields": ["content_no_index"],
"limit": FM_ROWS,
"consistencyLevel": "Strong",
"searchParams": {"metricType": "L2"},
},
timeout=180,
)
assert rsp["code"] == 0, rsp
rows = rsp.get("data", [])
ids = [row["id"] for row in rows]
assert len(ids) == len(set(ids)), rsp
distances = [row["distance"] for row in rows]
assert distances == sorted(distances), rsp
for row in rows:
assert row.get("content_no_index") == self._content_for_id(row["id"]), row
return sorted(ids)
@pytest.mark.tags(CaseLabel.L0)
def test_restful_fmindex_anchored_like(self):
"""
target: verify RESTful v2 can build and query a real FMINDEX
method: insert 2048 rows, build FMINDEX plus FLAT, then compare prefix/infix/suffix result PKs
expected: all three anchored LIKE forms return the exact expected rows
"""
name = self._create_collection("anchored_like")
self._insert_rows(name)
self._index_and_load(name)
expected = {
'content like "sta%"': [i for i in range(FM_ROWS) if i % 500 == 0],
'content like "%ool%"': [i for i in range(FM_ROWS) if i % 500 == 1],
'content like "%ary"': [i for i in range(FM_ROWS) if i % 500 == 2],
}
for expr, ids in expected.items():
assert self._query_ids(name, expr) == ids, expr
assert self._search_ids(name, expr) == ids, expr
twin_expr = expr.replace("content", "content_no_index", 1)
assert self._query_ids(name, twin_expr) == ids, twin_expr
@pytest.mark.tags(CaseLabel.L1)
def test_restful_fmindex_invalid_params(self):
"""
target: verify RESTful v2 returns server-side FMINDEX parameter errors
method: create a VARCHAR collection and submit invalid sample/block parameters
expected: both requests fail with the corresponding validation message
"""
name = self._create_collection("invalid_params")
cases = [
("invalid_sample_rate", {"fm_sa_sample_rate": 257}, "fm_sa_sample_rate for FM-index must be in"),
("invalid_block_bytes", {"fm_block_bytes": 24}, "fm_block_bytes for FM-index must be a power of two"),
]
for case_name, params, message in cases:
rsp = self.index_client.index_create(
{
"collectionName": name,
"indexParams": [
{
"fieldName": "content",
"indexName": f"bad_{case_name}",
"indexType": "FMINDEX",
"params": params,
}
],
}
)
assert rsp["code"] != 0, rsp
assert message in rsp.get("message", ""), rsp
@pytest.mark.tags(CaseLabel.L1)
@pytest.mark.parametrize(
"expression",
["membership_match(creator_id, {bf}, type=bloom)", "membership_match(creator_id, {rb}, type=roaring)"],
)
def test_restful_membership_blob_parameter_rejected(self, expression):
"""
target: verify JSON strings cannot masquerade as membership bytes and missing params fail at parsing
method: send a string template value, then omit the required expression parameter on a schema-only collection
expected: the first fails with a bytes type error and the second names the missing placeholder
"""
name = self._create_collection("membership_param", include_content=False)
param_key = "bf" if "bloom" in expression else "rb"
rsp = self.vector_client.vector_query(
{
"collectionName": name,
"filter": expression,
"exprParams": {param_key: "not-a-bytes-blob"},
"outputFields": ["id"],
}
)
assert rsp["code"] != 0, rsp
assert "bytes" in rsp.get("message", "").lower(), rsp
rsp = self.vector_client.vector_query({"collectionName": name, "filter": expression, "outputFields": ["id"]})
assert rsp["code"] != 0, rsp
assert f"expression template variable name {{{param_key}}} is not found" in rsp.get("message", ""), rsp
@pytest.mark.tags(CaseLabel.L1)
@pytest.mark.parametrize(
("expression", "param_key"),
[("bloom_match(creator_id, {bf})", "bf"), ("roaring_match(creator_id, {rb})", "rb")],
)
def test_restful_membership_predecessor_names_rejected(self, expression, param_key):
"""
target: verify RESTful v2 exposes only the unified membership_match name
method: submit each predecessor expression with a placeholder value
expected: parsing fails and directs the caller to membership_match
"""
name = self._create_collection("predecessor_name", include_content=False)
rsp = self.vector_client.vector_query(
{
"collectionName": name,
"filter": expression,
"exprParams": {param_key: "not-a-bytes-blob"},
"outputFields": ["id"],
}
)
assert rsp["code"] != 0, rsp
message = rsp.get("message", "")
assert "is not supported" in message, rsp
assert "membership_match" in message, rsp
class TestRestMembershipTransport:
@pytest.mark.tags(CaseLabel.L1)
def test_restful_membership_python_bytes_not_json_serializable(self):
"""
target: verify the REST test transport cannot encode a binary template value
method: pass Python bytes through VectorClient without contacting the server
expected: JSON serialization raises TypeError instead of altering the bytes
"""
client = VectorClient("http://serialization.invalid", "token")
with pytest.raises(TypeError, match="bytes"):
client.vector_query(
{
"collectionName": "serialization_is_local",
"filter": "membership_match(creator_id, {bf}, type=bloom)",
"exprParams": {"bf": b"binary"},
"outputFields": ["id"],
}
)