build(deps): bump github.com/blevesearch/bleve/v2 from 2.5.5 to 2.5.7

Bumps [github.com/blevesearch/bleve/v2](https://github.com/blevesearch/bleve) from 2.5.5 to 2.5.7.
- [Release notes](https://github.com/blevesearch/bleve/releases)
- [Commits](https://github.com/blevesearch/bleve/compare/v2.5.5...v2.5.7)

---
updated-dependencies:
- dependency-name: github.com/blevesearch/bleve/v2
  dependency-version: 2.5.7
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
This commit is contained in:
dependabot[bot]
2026-01-15 10:20:48 +00:00
committed by Ralf Haferkamp
parent e33ff722f7
commit 21207eba40
35 changed files with 1428 additions and 793 deletions

4
go.mod
View File

@@ -11,7 +11,7 @@ require (
github.com/Nerzal/gocloak/v13 v13.9.0
github.com/bbalet/stopwords v1.0.0
github.com/beevik/etree v1.6.0
github.com/blevesearch/bleve/v2 v2.5.5
github.com/blevesearch/bleve/v2 v2.5.7
github.com/cenkalti/backoff v2.2.1+incompatible
github.com/coreos/go-oidc/v3 v3.17.0
github.com/cs3org/go-cs3apis v0.0.0-20250908152307-4ca807afe54e
@@ -157,7 +157,7 @@ require (
github.com/blevesearch/zapx/v13 v13.4.2 // indirect
github.com/blevesearch/zapx/v14 v14.4.2 // indirect
github.com/blevesearch/zapx/v15 v15.4.2 // indirect
github.com/blevesearch/zapx/v16 v16.2.7 // indirect
github.com/blevesearch/zapx/v16 v16.2.8 // indirect
github.com/bluele/gcache v0.0.2 // indirect
github.com/bombsimon/logrusr/v3 v3.1.0 // indirect
github.com/cenkalti/backoff/v4 v4.3.0 // indirect

8
go.sum
View File

@@ -151,8 +151,8 @@ github.com/bits-and-blooms/bitset v1.12.0/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6
github.com/bits-and-blooms/bitset v1.22.0 h1:Tquv9S8+SGaS3EhyA+up3FXzmkhxPGjQQCkcs2uw7w4=
github.com/bits-and-blooms/bitset v1.22.0/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8=
github.com/bketelsen/crypt v0.0.3-0.20200106085610-5cbc8cc4026c/go.mod h1:MKsuJmJgSg28kpZDP6UIiPt0e0Oz0kqKNGyRaWEPv84=
github.com/blevesearch/bleve/v2 v2.5.5 h1:lzC89QUCco+y1qBnJxGqm4AbtsdsnlUvq0kXok8n3C8=
github.com/blevesearch/bleve/v2 v2.5.5/go.mod h1:t5WoESS5TDteTdnjhhvpA1BpLYErOBX2IQViTMLK7wo=
github.com/blevesearch/bleve/v2 v2.5.7 h1:2d9YrL5zrX5EBBW++GOaEKjE+NPWeZGaX77IM26m1Z8=
github.com/blevesearch/bleve/v2 v2.5.7/go.mod h1:yj0NlS7ocGC4VOSAedqDDMktdh2935v2CSWOCDMHdSA=
github.com/blevesearch/bleve_index_api v1.2.11 h1:bXQ54kVuwP8hdrXUSOnvTQfgK0KI1+f9A0ITJT8tX1s=
github.com/blevesearch/bleve_index_api v1.2.11/go.mod h1:rKQDl4u51uwafZxFrPD1R7xFOwKnzZW7s/LSeK4lgo0=
github.com/blevesearch/geo v0.2.4 h1:ECIGQhw+QALCZaDcogRTNSJYQXRtC8/m8IKiA706cqk=
@@ -185,8 +185,8 @@ github.com/blevesearch/zapx/v14 v14.4.2 h1:2SGHakVKd+TrtEqpfeq8X+So5PShQ5nW6GNxT
github.com/blevesearch/zapx/v14 v14.4.2/go.mod h1:rz0XNb/OZSMjNorufDGSpFpjoFKhXmppH9Hi7a877D8=
github.com/blevesearch/zapx/v15 v15.4.2 h1:sWxpDE0QQOTjyxYbAVjt3+0ieu8NCE0fDRaFxEsp31k=
github.com/blevesearch/zapx/v15 v15.4.2/go.mod h1:1pssev/59FsuWcgSnTa0OeEpOzmhtmr/0/11H0Z8+Nw=
github.com/blevesearch/zapx/v16 v16.2.7 h1:xcgFRa7f/tQXOwApVq7JWgPYSlzyUMmkuYa54tMDuR0=
github.com/blevesearch/zapx/v16 v16.2.7/go.mod h1:murSoCJPCk25MqURrcJaBQ1RekuqSCSfMjXH4rHyA14=
github.com/blevesearch/zapx/v16 v16.2.8 h1:SlnzF0YGtSlrsOE3oE7EgEX6BIepGpeqxs1IjMbHLQI=
github.com/blevesearch/zapx/v16 v16.2.8/go.mod h1:murSoCJPCk25MqURrcJaBQ1RekuqSCSfMjXH4rHyA14=
github.com/bluele/gcache v0.0.2 h1:WcbfdXICg7G/DGBh1PFfcirkWOQV+v077yF1pSy3DGw=
github.com/bluele/gcache v0.0.2/go.mod h1:m15KV+ECjptwSPxKhOhQoAFQVtUFjTVkc3H8o0t/fp0=
github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY=

View File

@@ -1,25 +0,0 @@
sudo: false
language: go
go:
- "1.21.x"
- "1.22.x"
- "1.23.x"
script:
- go get golang.org/x/tools/cmd/cover
- go get github.com/mattn/goveralls
- go get github.com/kisielk/errcheck
- go get -u github.com/FiloSottile/gvt
- gvt restore
- go test -race -v $(go list ./... | grep -v vendor/)
- go vet $(go list ./... | grep -v vendor/)
- go test ./test -v -indexType scorch
- errcheck -ignorepkg fmt $(go list ./... | grep -v vendor/);
- scripts/project-code-coverage.sh
- scripts/build_children.sh
notifications:
email:
- fts-team@couchbase.com

View File

@@ -1,7 +1,7 @@
# ![bleve](docs/bleve.png) bleve
[![Tests](https://github.com/blevesearch/bleve/actions/workflows/tests.yml/badge.svg?branch=master&event=push)](https://github.com/blevesearch/bleve/actions/workflows/tests.yml?query=event%3Apush+branch%3Amaster)
[![Coverage Status](https://coveralls.io/repos/github/blevesearch/bleve/badge.svg?branch=master)](https://coveralls.io/github/blevesearch/bleve?branch=master)
[![Coverage Status](https://coveralls.io/repos/github/blevesearch/bleve/badge.svg)](https://coveralls.io/github/blevesearch/bleve)
[![Go Reference](https://pkg.go.dev/badge/github.com/blevesearch/bleve/v2.svg)](https://pkg.go.dev/github.com/blevesearch/bleve/v2)
[![Join the chat](https://badges.gitter.im/join_chat.svg)](https://app.gitter.im/#/room/#blevesearch_bleve:gitter.im)
[![Go Report Card](https://goreportcard.com/badge/github.com/blevesearch/bleve/v2)](https://goreportcard.com/report/github.com/blevesearch/bleve/v2)

View File

@@ -180,7 +180,7 @@ func NewGeoShapeFieldFromShapeWithIndexingOptions(name string, arrayPositions []
// docvalues are always enabled for geoshape fields, even if the
// indexing options are set to not include docvalues.
options = options | index.DocValues
options |= index.DocValues
return &GeoShapeField{
shape: shape,
@@ -232,7 +232,7 @@ func NewGeometryCollectionFieldFromShapesWithIndexingOptions(name string,
// docvalues are always enabled for geoshape fields, even if the
// indexing options are set to not include docvalues.
options = options | index.DocValues
options |= index.DocValues
return &GeoShapeField{
shape: shape,

View File

@@ -109,6 +109,10 @@ func NewVectorField(name string, arrayPositions []uint64,
func NewVectorFieldWithIndexingOptions(name string, arrayPositions []uint64,
vector []float32, dims int, similarity, vectorIndexOptimizedFor string,
options index.FieldIndexingOptions) *VectorField {
// ensure the options are set to not store/index term vectors/doc values
options &^= index.StoreField | index.IncludeTermVectors | index.DocValues
// skip freq/norms for vector field
options |= index.SkipFreqNorm
return &VectorField{
name: name,

View File

@@ -17,113 +17,125 @@ package fusion
import (
"fmt"
"sort"
"github.com/blevesearch/bleve/v2/search"
)
// formatRRFMessage builds the explanation string for a single component of the
// Reciprocal Rank Fusion calculation.
func formatRRFMessage(weight float64, rank int, rankConstant int) string {
return fmt.Sprintf("rrf score (weight=%.3f, rank=%d, rank_constant=%d), normalized score of", weight, rank, rankConstant)
}
// ReciprocalRankFusion performs a reciprocal rank fusion on the search results.
func ReciprocalRankFusion(hits search.DocumentMatchCollection, weights []float64, rankConstant int, windowSize int, numKNNQueries int, explain bool) FusionResult {
if len(hits) == 0 {
return FusionResult{
Hits: hits,
// ReciprocalRankFusion applies Reciprocal Rank Fusion across the primary FTS
// results and each KNN sub-query. Ranks are limited to `windowSize` per source,
// weighted, and combined into a single fused score, with optional explanation
// details.
func ReciprocalRankFusion(hits search.DocumentMatchCollection, weights []float64, rankConstant int, windowSize int, numKNNQueries int, explain bool) *FusionResult {
nHits := len(hits)
if nHits == 0 || windowSize == 0 {
return &FusionResult{
Hits: search.DocumentMatchCollection{},
Total: 0,
MaxScore: 0.0,
}
}
// Create a map of document ID to a slice of ranks.
// The first element of the slice is the rank from the FTS search,
// and the subsequent elements are the ranks from the KNN searches.
docRanks := make(map[string][]int)
limit := min(nHits, windowSize)
// Pre-assign rank lists to each candidate document
for _, hit := range hits {
docRanks[hit.ID] = make([]int, numKNNQueries+1)
// precompute rank+scores to prevent additional division ops later
rankReciprocals := make([]float64, limit)
for i := range rankReciprocals {
rankReciprocals[i] = 1.0 / float64(rankConstant+i+1)
}
// Only a max of `window_size` elements need to be counted for. Stop
// calculating rank once this threshold is hit.
sort.Slice(hits, func(a, b int) bool {
return scoreSortFunc()(hits[a], hits[b]) < 0
})
// Only consider top windowSize docs for rescoring
for i := range min(windowSize, len(hits)) {
if hits[i].Score != 0.0 {
// Skip if Score is 0, since that means the document was not
// found as part of FTS, and only in KNN.
docRanks[hits[i].ID][0] = i + 1
// init explanations if required
var fusionExpl map[*search.DocumentMatch][]*search.Explanation
if explain {
fusionExpl = make(map[*search.DocumentMatch][]*search.Explanation, nHits)
}
// The code here mainly deals with obtaining rank/score for fts hits.
// First sort hits by score
sortDocMatchesByScore(hits)
// Calculate fts rank+scores
ftsWeight := weights[0]
for i := 0; i < nHits; i++ {
if i < windowSize {
hit := hits[i]
// No fts scores from this hit onwards, break loop
if hit.Score == 0.0 {
break
}
contrib := ftsWeight * rankReciprocals[i]
hit.Score = contrib
if explain {
expl := getFusionExplAt(
hit,
0,
contrib,
formatRRFMessage(ftsWeight, i+1, rankConstant),
)
fusionExpl[hit] = append(fusionExpl[hit], expl)
}
} else {
// These FTS hits are not counted in the results, so set to 0
hits[i].Score = 0.0
}
}
// Allocate knnDocs and reuse it within the loop
knnDocs := make([]*search.DocumentMatch, 0, len(hits))
// Code from here is to calculate knn ranks and scores
// iterate over each knn query and calculate knn rank+scores
for queryIdx := 0; queryIdx < numKNNQueries; queryIdx++ {
knnWeight := weights[queryIdx+1]
// Sorts hits in decreasing order of hit.ScoreBreakdown[i]
sortDocMatchesByBreakdown(hits, queryIdx)
// For each KNN query, rank the documents based on their KNN score.
for i := range numKNNQueries {
knnDocs = knnDocs[:0]
for i := 0; i < nHits; i++ {
// break if score breakdown doesn't exist (sort function puts these hits at the end)
// or if we go past the windowSize
_, scoreBreakdownExists := scoreBreakdownForQuery(hits[i], queryIdx)
if i >= windowSize || !scoreBreakdownExists {
break
}
for _, hit := range hits {
if _, ok := hit.ScoreBreakdown[i]; ok {
knnDocs = append(knnDocs, hit)
hit := hits[i]
contrib := knnWeight * rankReciprocals[i]
hit.Score += contrib
if explain {
expl := getFusionExplAt(
hit,
queryIdx+1,
contrib,
formatRRFMessage(knnWeight, i+1, rankConstant),
)
fusionExpl[hit] = append(fusionExpl[hit], expl)
}
}
// Sort the documents based on their score for this KNN query.
sort.Slice(knnDocs, func(a, b int) bool {
return scoreBreakdownSortFunc(i)(knnDocs[a], knnDocs[b]) < 0
})
// Update the ranks of the documents in the docRanks map.
// Only consider top windowSize docs for rescoring.
for j := range min(windowSize, len(knnDocs)) {
docRanks[knnDocs[j].ID][i+1] = j + 1
}
}
// Calculate the RRF score for each document.
var maxScore float64
for _, hit := range hits {
var rrfScore float64
var explChildren []*search.Explanation
if explain {
explChildren = make([]*search.Explanation, 0, numKNNQueries+1)
finalizeFusionExpl(hit, fusionExpl[hit])
}
for i, rank := range docRanks[hit.ID] {
if rank > 0 {
partialRrfScore := weights[i] * 1.0 / float64(rankConstant+rank)
if explain {
expl := getFusionExplAt(
hit,
i,
partialRrfScore,
formatRRFMessage(weights[i], rank, rankConstant),
)
explChildren = append(explChildren, expl)
}
rrfScore += partialRrfScore
}
}
hit.Score = rrfScore
hit.ScoreBreakdown = nil
if rrfScore > maxScore {
maxScore = rrfScore
}
if explain {
finalizeFusionExpl(hit, explChildren)
if hit.Score > maxScore {
maxScore = hit.Score
}
}
sort.Sort(hits)
if len(hits) > windowSize {
sortDocMatchesByScore(hits)
if nHits > windowSize {
hits = hits[:windowSize]
}
return FusionResult{
return &FusionResult{
Hits: hits,
Total: uint64(len(hits)),
MaxScore: maxScore,

View File

@@ -16,145 +16,147 @@ package fusion
import (
"fmt"
"sort"
"github.com/blevesearch/bleve/v2/search"
)
// formatRSFMessage builds the explanation string associated with a single
// component of the Relative Score Fusion calculation.
func formatRSFMessage(weight float64, normalizedScore float64, minScore float64, maxScore float64) string {
return fmt.Sprintf("rsf score (weight=%.3f, normalized=%.6f, min=%.6f, max=%.6f), normalized score of",
weight, normalizedScore, minScore, maxScore)
}
// RelativeScoreFusion normalizes scores based on min/max values for FTS and each KNN query, then applies weights.
func RelativeScoreFusion(hits search.DocumentMatchCollection, weights []float64, windowSize int, numKNNQueries int, explain bool) FusionResult {
if len(hits) == 0 {
return FusionResult{
Hits: hits,
// RelativeScoreFusion normalizes the best-scoring documents from the primary
// FTS query and each KNN query, scales those normalized values by the supplied
// weights, and combines them into a single fused score. Only the top
// `windowSize` documents per source are considered, and explanations are
// materialized lazily when requested.
func RelativeScoreFusion(hits search.DocumentMatchCollection, weights []float64, windowSize int, numKNNQueries int, explain bool) *FusionResult {
nHits := len(hits)
if nHits == 0 || windowSize == 0 {
return &FusionResult{
Hits: search.DocumentMatchCollection{},
Total: 0,
MaxScore: 0.0,
}
}
rsfScores := make(map[string]float64)
// contains the docs under consideration for scoring.
// Reused for fts and knn hits
scoringDocs := make([]*search.DocumentMatch, 0, len(hits))
var explMap map[string][]*search.Explanation
// init explanations if required
var fusionExpl map[*search.DocumentMatch][]*search.Explanation
if explain {
explMap = make(map[string][]*search.Explanation)
fusionExpl = make(map[*search.DocumentMatch][]*search.Explanation, nHits)
}
// remove non-fts hits
// Code here for calculating fts results
// Sort by fts scores
sortDocMatchesByScore(hits)
// ftsLimit holds the total number of fts hits to consider for rsf
ftsLimit := 0
for _, hit := range hits {
if hit.Score != 0.0 {
scoringDocs = append(scoringDocs, hit)
if hit.Score == 0.0 {
break
}
ftsLimit++
}
// sort hits by fts score
sort.Slice(scoringDocs, func(a, b int) bool {
return scoreSortFunc()(scoringDocs[a], scoringDocs[b]) < 0
})
// Reslice to correct size
if len(scoringDocs) > windowSize {
scoringDocs = scoringDocs[:windowSize]
}
ftsLimit = min(ftsLimit, windowSize)
var min, max float64
if len(scoringDocs) > 0 {
min, max = scoringDocs[len(scoringDocs)-1].Score, scoringDocs[0].Score
}
// calculate fts scores
if ftsLimit > 0 {
max := hits[0].Score
min := hits[ftsLimit-1].Score
denom := max - min
weight := weights[0]
for _, hit := range scoringDocs {
var tempRsfScore float64
if max > min {
tempRsfScore = (hit.Score - min) / (max - min)
} else {
tempRsfScore = 1.0
}
if explain {
// create and replace new explanation
expl := getFusionExplAt(
hit,
0,
tempRsfScore,
formatRSFMessage(weights[0], tempRsfScore, min, max),
)
explMap[hit.ID] = append(explMap[hit.ID], expl)
}
rsfScores[hit.ID] = weights[0] * tempRsfScore
}
for i := range numKNNQueries {
scoringDocs = scoringDocs[:0]
for _, hit := range hits {
if _, exists := hit.ScoreBreakdown[i]; exists {
scoringDocs = append(scoringDocs, hit)
for i := 0; i < ftsLimit; i++ {
hit := hits[i]
norm := 1.0
if denom > 0 {
norm = (hit.Score - min) / denom
}
}
sort.Slice(scoringDocs, func(a, b int) bool {
return scoreBreakdownSortFunc(i)(scoringDocs[a], scoringDocs[b]) < 0
})
if len(scoringDocs) > windowSize {
scoringDocs = scoringDocs[:windowSize]
}
if len(scoringDocs) > 0 {
min, max = scoringDocs[len(scoringDocs)-1].ScoreBreakdown[i], scoringDocs[0].ScoreBreakdown[i]
} else {
min, max = 0.0, 0.0
}
for _, hit := range scoringDocs {
var tempRsfScore float64
if max > min {
tempRsfScore = (hit.ScoreBreakdown[i] - min) / (max - min)
} else {
tempRsfScore = 1.0
}
contrib := weight * norm
if explain {
expl := getFusionExplAt(
hit,
i+1,
tempRsfScore,
formatRSFMessage(weights[i+1], tempRsfScore, min, max),
0,
norm,
formatRSFMessage(weight, norm, min, max),
)
explMap[hit.ID] = append(explMap[hit.ID], expl)
fusionExpl[hit] = append(fusionExpl[hit], expl)
}
rsfScores[hit.ID] += weights[i+1] * tempRsfScore
hit.Score = contrib
}
for i := ftsLimit; i < nHits; i++ {
// These FTS hits are not counted in the results, so set to 0
hits[i].Score = 0.0
}
}
var maxScore float64
for _, hit := range hits {
if rsfScore, exists := rsfScores[hit.ID]; exists {
hit.Score = rsfScore
if rsfScore > maxScore {
maxScore = rsfScore
// Code from here is for calculating knn scores
for queryIdx := 0; queryIdx < numKNNQueries; queryIdx++ {
sortDocMatchesByBreakdown(hits, queryIdx)
// knnLimit holds the total number of knn hits retrieved for a specific knn query
knnLimit := 0
for _, hit := range hits {
if _, ok := scoreBreakdownForQuery(hit, queryIdx); !ok {
break
}
if explain {
finalizeFusionExpl(hit, explMap[hit.ID])
}
} else {
hit.Score = 0.0
knnLimit++
}
knnLimit = min(knnLimit, windowSize)
// if limit is 0, skip calculating
if knnLimit == 0 {
continue
}
max, _ := scoreBreakdownForQuery(hits[0], queryIdx)
min, _ := scoreBreakdownForQuery(hits[knnLimit-1], queryIdx)
denom := max - min
weight := weights[queryIdx+1]
for i := 0; i < knnLimit; i++ {
hit := hits[i]
score, _ := scoreBreakdownForQuery(hit, queryIdx)
norm := 1.0
if denom > 0 {
norm = (score - min) / denom
}
contrib := weight * norm
if explain {
expl := getFusionExplAt(
hit,
queryIdx+1,
norm,
formatRSFMessage(weight, norm, min, max),
)
fusionExpl[hit] = append(fusionExpl[hit], expl)
}
hit.Score += contrib
}
}
// Finalize scores
var maxScore float64
for _, hit := range hits {
if explain {
finalizeFusionExpl(hit, fusionExpl[hit])
}
if hit.Score > maxScore {
maxScore = hit.Score
}
hit.ScoreBreakdown = nil
}
sort.Sort(hits)
sortDocMatchesByScore(hits)
if len(hits) > windowSize {
if nHits > windowSize {
hits = hits[:windowSize]
}
return FusionResult{
return &FusionResult{
Hits: hits,
Total: uint64(len(hits)),
MaxScore: maxScore,

View File

@@ -16,70 +16,82 @@
package fusion
import (
"sort"
"github.com/blevesearch/bleve/v2/search"
)
// scoreBreakdownSortFunc returns a comparison function for sorting DocumentMatch objects
// by their ScoreBreakdown at the specified index in descending order.
// In case of ties, documents with lower HitNumber (earlier hits) are preferred.
// If either document is missing the ScoreBreakdown for the specified index,
// it's treated as having a score of 0.0.
func scoreBreakdownSortFunc(idx int) func(i, j *search.DocumentMatch) int {
return func(i, j *search.DocumentMatch) int {
// Safely extract scores, defaulting to 0.0 if missing
iScore := 0.0
jScore := 0.0
if i.ScoreBreakdown != nil {
if score, ok := i.ScoreBreakdown[idx]; ok {
iScore = score
}
}
if j.ScoreBreakdown != nil {
if score, ok := j.ScoreBreakdown[idx]; ok {
jScore = score
}
}
// Sort by score in descending order (higher scores first)
if iScore > jScore {
return -1
} else if iScore < jScore {
return 1
}
// Break ties by HitNumber in ascending order (lower HitNumber wins)
if i.HitNumber < j.HitNumber {
return -1
} else if i.HitNumber > j.HitNumber {
return 1
}
return 0 // Equal scores and HitNumbers
// sortDocMatchesByScore orders the provided collection in-place by the primary
// score in descending order, breaking ties with the original `HitNumber` to
// ensure deterministic output.
func sortDocMatchesByScore(hits search.DocumentMatchCollection) {
if len(hits) < 2 {
return
}
sort.Slice(hits, func(a, b int) bool {
i := hits[a]
j := hits[b]
if i.Score == j.Score {
return i.HitNumber < j.HitNumber
}
return i.Score > j.Score
})
}
func scoreSortFunc() func(i, j *search.DocumentMatch) int {
return func(i, j *search.DocumentMatch) int {
// Sort by score in descending order
if i.Score > j.Score {
return -1
} else if i.Score < j.Score {
return 1
}
// Break ties by HitNumber
if i.HitNumber < j.HitNumber {
return -1
} else if i.HitNumber > j.HitNumber {
return 1
}
return 0
// scoreBreakdownForQuery fetches the score for a specific KNN query index from
// the provided hit. The boolean return indicates whether the score is present.
func scoreBreakdownForQuery(hit *search.DocumentMatch, idx int) (float64, bool) {
if hit == nil || hit.ScoreBreakdown == nil {
return 0, false
}
score, ok := hit.ScoreBreakdown[idx]
return score, ok
}
// sortDocMatchesByBreakdown orders the hits in-place using the KNN score for
// the supplied query index (descending), breaking ties with `HitNumber` and
// placing hits without a score at the end.
func sortDocMatchesByBreakdown(hits search.DocumentMatchCollection, queryIdx int) {
if len(hits) < 2 {
return
}
sort.SliceStable(hits, func(a, b int) bool {
left := hits[a]
right := hits[b]
var leftScore float64
leftOK := false
if left != nil && left.ScoreBreakdown != nil {
leftScore, leftOK = left.ScoreBreakdown[queryIdx]
}
var rightScore float64
rightOK := false
if right != nil && right.ScoreBreakdown != nil {
rightScore, rightOK = right.ScoreBreakdown[queryIdx]
}
if leftOK && rightOK {
if leftScore == rightScore {
return left.HitNumber < right.HitNumber
}
return leftScore > rightScore
}
if leftOK != rightOK {
return leftOK
}
return left.HitNumber < right.HitNumber
})
}
// getFusionExplAt copies the existing explanation child at the requested index
// and wraps it in a new node describing how the fusion algorithm adjusted the
// score.
func getFusionExplAt(hit *search.DocumentMatch, i int, value float64, message string) *search.Explanation {
return &search.Explanation{
Value: value,
@@ -88,6 +100,9 @@ func getFusionExplAt(hit *search.DocumentMatch, i int, value float64, message st
}
}
// finalizeFusionExpl installs the collection of fusion explanation children and
// updates the root message so the caller sees the fused score as the sum of its
// parts.
func finalizeFusionExpl(hit *search.DocumentMatch, explChildren []*search.Explanation) {
hit.Expl.Children = explChildren

View File

@@ -35,43 +35,45 @@ type Event struct {
// EventKind represents an event code for OnEvent() callbacks.
type EventKind int
// EventKindCloseStart is fired when a Scorch.Close() has begun.
var EventKindCloseStart = EventKind(1)
const (
// EventKindCloseStart is fired when a Scorch.Close() has begun.
EventKindCloseStart EventKind = iota
// EventKindClose is fired when a scorch index has been fully closed.
var EventKindClose = EventKind(2)
// EventKindClose is fired when a scorch index has been fully closed.
EventKindClose
// EventKindMergerProgress is fired when the merger has completed a
// round of merge processing.
var EventKindMergerProgress = EventKind(3)
// EventKindMergerProgress is fired when the merger has completed a
// round of merge processing.
EventKindMergerProgress
// EventKindPersisterProgress is fired when the persister has completed
// a round of persistence processing.
var EventKindPersisterProgress = EventKind(4)
// EventKindPersisterProgress is fired when the persister has completed
// a round of persistence processing.
EventKindPersisterProgress
// EventKindBatchIntroductionStart is fired when Batch() is invoked which
// introduces a new segment.
var EventKindBatchIntroductionStart = EventKind(5)
// EventKindBatchIntroductionStart is fired when Batch() is invoked which
// introduces a new segment.
EventKindBatchIntroductionStart
// EventKindBatchIntroduction is fired when Batch() completes.
var EventKindBatchIntroduction = EventKind(6)
// EventKindBatchIntroduction is fired when Batch() completes.
EventKindBatchIntroduction
// EventKindMergeTaskIntroductionStart is fired when the merger is about to
// start the introduction of merged segment from a single merge task.
var EventKindMergeTaskIntroductionStart = EventKind(7)
// EventKindMergeTaskIntroductionStart is fired when the merger is about to
// start the introduction of merged segment from a single merge task.
EventKindMergeTaskIntroductionStart
// EventKindMergeTaskIntroduction is fired when the merger has completed
// the introduction of merged segment from a single merge task.
var EventKindMergeTaskIntroduction = EventKind(8)
// EventKindMergeTaskIntroduction is fired when the merger has completed
// the introduction of merged segment from a single merge task.
EventKindMergeTaskIntroduction
// EventKindPreMergeCheck is fired before the merge begins to check if
// the caller should proceed with the merge.
var EventKindPreMergeCheck = EventKind(9)
// EventKindPreMergeCheck is fired before the merge begins to check if
// the caller should proceed with the merge.
EventKindPreMergeCheck
// EventKindIndexStart is fired when Index() is invoked which
// creates a new Document object from an interface using the index mapping.
var EventKindIndexStart = EventKind(10)
// EventKindIndexStart is fired when Index() is invoked which
// creates a new Document object from an interface using the index mapping.
EventKindIndexStart
// EventKindPurgerCheck is fired before the purge code is invoked and decides
// whether to execute or not. For unit test purposes
var EventKindPurgerCheck = EventKind(11)
// EventKindPurgerCheck is fired before the purge code is invoked and decides
// whether to execute or not. For unit test purposes
EventKindPurgerCheck
)

View File

@@ -24,6 +24,8 @@ import (
segment "github.com/blevesearch/scorch_segment_api/v2"
)
const introducer = "introducer"
type segmentIntroduction struct {
id uint64
data segment.Segment
@@ -50,10 +52,11 @@ type epochWatcher struct {
func (s *Scorch) introducerLoop() {
defer func() {
if r := recover(); r != nil {
s.fireAsyncError(&AsyncPanicError{
Source: "introducer",
Path: s.path,
})
s.fireAsyncError(NewScorchError(
introducer,
fmt.Sprintf("panic: %v, path: %s", r, s.path),
ErrAsyncPanic,
))
}
s.asyncTasks.Done()

View File

@@ -29,13 +29,16 @@ import (
segment "github.com/blevesearch/scorch_segment_api/v2"
)
const merger = "merger"
func (s *Scorch) mergerLoop() {
defer func() {
if r := recover(); r != nil {
s.fireAsyncError(&AsyncPanicError{
Source: "merger",
Path: s.path,
})
s.fireAsyncError(NewScorchError(
merger,
fmt.Sprintf("panic: %v, path: %s", r, s.path),
ErrAsyncPanic,
))
}
s.asyncTasks.Done()
@@ -45,7 +48,11 @@ func (s *Scorch) mergerLoop() {
var ctrlMsg *mergerCtrl
mergePlannerOptions, err := s.parseMergePlannerOptions()
if err != nil {
s.fireAsyncError(fmt.Errorf("mergePlannerOption json parsing err: %v", err))
s.fireAsyncError(NewScorchError(
merger,
fmt.Sprintf("mergerPlannerOptions json parsing err: %v", err),
ErrOptionsParse,
))
return
}
ctrlMsgDflt := &mergerCtrl{ctx: context.Background(),
@@ -110,7 +117,12 @@ OUTER:
ctrlMsg = nil
break OUTER
}
s.fireAsyncError(fmt.Errorf("merging err: %v", err))
s.fireAsyncError(NewScorchError(
merger,
fmt.Sprintf("merging err: %v", err),
ErrPersist,
))
_ = ourSnapshot.DecRef()
atomic.AddUint64(&s.stats.TotFileMergeLoopErr, 1)
continue OUTER

View File

@@ -38,6 +38,8 @@ import (
bolt "go.etcd.io/bbolt"
)
const persister = "persister"
// DefaultPersisterNapTimeMSec is kept to zero as this helps in direct
// persistence of segments with the default safe batch option.
// If the default safe batch option results in high number of
@@ -95,10 +97,11 @@ type notificationChan chan struct{}
func (s *Scorch) persisterLoop() {
defer func() {
if r := recover(); r != nil {
s.fireAsyncError(&AsyncPanicError{
Source: "persister",
Path: s.path,
})
s.fireAsyncError(NewScorchError(
persister,
fmt.Sprintf("panic: %v, path: %s", r, s.path),
ErrAsyncPanic,
))
}
s.asyncTasks.Done()
@@ -112,7 +115,11 @@ func (s *Scorch) persisterLoop() {
po, err := s.parsePersisterOptions()
if err != nil {
s.fireAsyncError(fmt.Errorf("persisterOptions json parsing err: %v", err))
s.fireAsyncError(NewScorchError(
persister,
fmt.Sprintf("persisterOptions json parsing err: %v", err),
ErrOptionsParse,
))
return
}
@@ -173,7 +180,11 @@ OUTER:
// the retry attempt
unpersistedCallbacks = append(unpersistedCallbacks, ourPersistedCallbacks...)
s.fireAsyncError(fmt.Errorf("got err persisting snapshot: %v", err))
s.fireAsyncError(NewScorchError(
persister,
fmt.Sprintf("got err persisting snapshot: %v", err),
ErrPersist,
))
_ = ourSnapshot.DecRef()
atomic.AddUint64(&s.stats.TotPersistLoopErr, 1)
continue OUTER
@@ -1060,13 +1071,21 @@ func (s *Scorch) loadSegment(segmentBucket *bolt.Bucket) (*SegmentSnapshot, erro
func (s *Scorch) removeOldData() {
removed, err := s.removeOldBoltSnapshots()
if err != nil {
s.fireAsyncError(fmt.Errorf("got err removing old bolt snapshots: %v", err))
s.fireAsyncError(NewScorchError(
persister,
fmt.Sprintf("got err removing old bolt snapshots: %v", err),
ErrCleanup,
))
}
atomic.AddUint64(&s.stats.TotSnapshotsRemovedFromMetaStore, uint64(removed))
err = s.removeOldZapFiles()
if err != nil {
s.fireAsyncError(fmt.Errorf("got err removing old zap files: %v", err))
s.fireAsyncError(NewScorchError(
persister,
fmt.Sprintf("got err removing old zap files: %v", err),
ErrCleanup,
))
}
}

View File

@@ -88,14 +88,45 @@ type Scorch struct {
spatialPlugin index.SpatialAnalyzerPlugin
}
// AsyncPanicError is passed to scorch asyncErrorHandler when panic occurs in scorch background process
type AsyncPanicError struct {
Source string
Path string
type ScorchErrorType string
func (t ScorchErrorType) Error() string {
return string(t)
}
func (e *AsyncPanicError) Error() string {
return fmt.Sprintf("%s panic when processing %s", e.Source, e.Path)
// ErrType values for ScorchError
const (
ErrAsyncPanic = ScorchErrorType("async panic error")
ErrPersist = ScorchErrorType("persist error")
ErrCleanup = ScorchErrorType("cleanup error")
ErrOptionsParse = ScorchErrorType("options parse error")
)
// ScorchError is passed to onAsyncError when errors are
// fired from scorch background processes
type ScorchError struct {
Source string
ErrMsg string
ErrType ScorchErrorType
}
func (e *ScorchError) Error() string {
return fmt.Sprintf("source: %s, %v: %s", e.Source, e.ErrType, e.ErrMsg)
}
// Lets the onAsyncError function verify what type of
// error is fired using errors.Is(...). This lets the function
// handle errors differently.
func (e *ScorchError) Unwrap() error {
return e.ErrType
}
func NewScorchError(source, errMsg string, errType ScorchErrorType) error {
return &ScorchError{
Source: source,
ErrMsg: errMsg,
ErrType: errType,
}
}
type internalStats struct {

View File

@@ -23,7 +23,6 @@ import (
"path/filepath"
"reflect"
"sort"
"strings"
"sync"
"sync/atomic"
@@ -147,7 +146,7 @@ func (is *IndexSnapshot) newIndexSnapshotFieldDict(field string,
makeItr func(i segment.TermDictionary) segment.DictionaryIterator,
randomLookup bool,
) (*IndexSnapshotFieldDict, error) {
results := make(chan *asynchSegmentResult)
results := make(chan *asynchSegmentResult, len(is.segment))
var totalBytesRead uint64
var fieldCardinality int64
for _, s := range is.segment {
@@ -281,10 +280,13 @@ func (is *IndexSnapshot) FieldDictRange(field string, startTerm []byte,
// to use as the end key in a traditional (inclusive, exclusive]
// start/end range
func calculateExclusiveEndFromPrefix(in []byte) []byte {
if len(in) == 0 {
return nil
}
rv := make([]byte, len(in))
copy(rv, in)
for i := len(rv) - 1; i >= 0; i-- {
rv[i] = rv[i] + 1
rv[i]++
if rv[i] != 0 {
return rv // didn't overflow, so stop
}
@@ -391,7 +393,7 @@ func (is *IndexSnapshot) FieldDictContains(field string) (index.FieldDictContain
}
func (is *IndexSnapshot) DocIDReaderAll() (index.DocIDReader, error) {
results := make(chan *asynchSegmentResult)
results := make(chan *asynchSegmentResult, len(is.segment))
for index, segment := range is.segment {
go func(index int, segment *SegmentSnapshot) {
results <- &asynchSegmentResult{
@@ -405,7 +407,7 @@ func (is *IndexSnapshot) DocIDReaderAll() (index.DocIDReader, error) {
}
func (is *IndexSnapshot) DocIDReaderOnly(ids []string) (index.DocIDReader, error) {
results := make(chan *asynchSegmentResult)
results := make(chan *asynchSegmentResult, len(is.segment))
for index, segment := range is.segment {
go func(index int, segment *SegmentSnapshot) {
docs, err := segment.DocNumbers(ids)
@@ -451,7 +453,7 @@ func (is *IndexSnapshot) newDocIDReader(results chan *asynchSegmentResult) (inde
func (is *IndexSnapshot) Fields() ([]string, error) {
// FIXME not making this concurrent for now as it's not used in hot path
// of any searches at the moment (just a debug aid)
fieldsMap := map[string]struct{}{}
fieldsMap := make(map[string]struct{})
for _, segment := range is.segment {
fields := segment.Fields()
for _, field := range fields {
@@ -765,7 +767,7 @@ func (is *IndexSnapshot) recycleTermFieldReader(tfr *IndexSnapshotTermFieldReade
is.m2.Lock()
if is.fieldTFRs == nil {
is.fieldTFRs = map[string][]*IndexSnapshotTermFieldReader{}
is.fieldTFRs = make(map[string][]*IndexSnapshotTermFieldReader)
}
if len(is.fieldTFRs[tfr.field]) < is.getFieldTFRCacheThreshold() {
tfr.bytesRead = 0
@@ -813,7 +815,7 @@ func (is *IndexSnapshot) documentVisitFieldTermsOnSegment(
// Filter out fields that have been completely deleted or had their
// docvalues data deleted from both visitable fields and required fields
filterUpdatedFields := func(fields []string) []string {
filteredFields := make([]string, 0)
filteredFields := make([]string, 0, len(fields))
for _, field := range fields {
if info, ok := is.updatedFields[field]; ok &&
(info.DocValues || info.Deleted) {
@@ -978,15 +980,17 @@ func subtractStrings(a, b []string) []string {
return a
}
// Create a map for O(1) lookups
bMap := make(map[string]struct{}, len(b))
for _, bs := range b {
bMap[bs] = struct{}{}
}
rv := make([]string, 0, len(a))
OUTER:
for _, as := range a {
for _, bs := range b {
if as == bs {
continue OUTER
}
if _, exists := bMap[as]; !exists {
rv = append(rv, as)
}
rv = append(rv, as)
}
return rv
}
@@ -1279,7 +1283,7 @@ func (is *IndexSnapshot) TermFrequencies(field string, limit int, descending boo
sort.Slice(termFreqs, func(i, j int) bool {
if termFreqs[i].Frequency == termFreqs[j].Frequency {
// If frequencies are equal, sort by term lexicographically
return strings.Compare(termFreqs[i].Term, termFreqs[j].Term) < 0
return termFreqs[i].Term < termFreqs[j].Term
}
if descending {
return termFreqs[i].Frequency > termFreqs[j].Frequency

View File

@@ -37,14 +37,10 @@ func (is *IndexSnapshot) VectorReader(ctx context.Context, vector []float32,
snapshot: is,
searchParams: searchParams,
eligibleSelector: eligibleSelector,
postings: make([]segment_api.VecPostingsList, len(is.segment)),
iterators: make([]segment_api.VecPostingsIterator, len(is.segment)),
}
if rv.postings == nil {
rv.postings = make([]segment_api.VecPostingsList, len(is.segment))
}
if rv.iterators == nil {
rv.iterators = make([]segment_api.VecPostingsIterator, len(is.segment))
}
// initialize postings and iterators within the OptimizeVR's Finish()
return rv, nil
}

View File

@@ -18,7 +18,6 @@ import (
"context"
"fmt"
"sort"
"strings"
"sync"
"time"
@@ -905,7 +904,7 @@ func preSearchDataSearch(ctx context.Context, req *SearchRequest, flags *preSear
// which would happen in the case of an alias tree and depending on the level of the tree, the preSearchData
// needs to be redistributed to the indexes at that level
func redistributePreSearchData(req *SearchRequest, indexes []Index) (map[string]map[string]interface{}, error) {
rv := make(map[string]map[string]interface{})
rv := make(map[string]map[string]interface{}, len(indexes))
for _, index := range indexes {
rv[index.Name()] = make(map[string]interface{})
}
@@ -1202,23 +1201,16 @@ func (i *indexAliasImpl) TermFrequencies(field string, limit int, descending boo
})
}
if descending {
sort.Slice(rvTermFreqs, func(i, j int) bool {
if rvTermFreqs[i].Frequency == rvTermFreqs[j].Frequency {
// If frequencies are equal, sort by term lexicographically
return strings.Compare(rvTermFreqs[i].Term, rvTermFreqs[j].Term) < 0
}
sort.Slice(rvTermFreqs, func(i, j int) bool {
if rvTermFreqs[i].Frequency == rvTermFreqs[j].Frequency {
// If frequencies are equal, sort by term lexicographically
return rvTermFreqs[i].Term < rvTermFreqs[j].Term
}
if descending {
return rvTermFreqs[i].Frequency > rvTermFreqs[j].Frequency
})
} else {
sort.Slice(rvTermFreqs, func(i, j int) bool {
if rvTermFreqs[i].Frequency == rvTermFreqs[j].Frequency {
// If frequencies are equal, sort by term lexicographically
return strings.Compare(rvTermFreqs[i].Term, rvTermFreqs[j].Term) < 0
}
return rvTermFreqs[i].Frequency < rvTermFreqs[j].Frequency
})
}
}
return rvTermFreqs[i].Frequency < rvTermFreqs[j].Frequency
})
if limit > len(rvTermFreqs) {
limit = len(rvTermFreqs)
@@ -1272,25 +1264,22 @@ func (i *indexAliasImpl) CentroidCardinalities(field string, limit int, descendi
close(asyncResults)
}()
rvCentroidCardinalitiesResult := make([]index.CentroidCardinality, 0, limit)
rvCentroidCardinalities := make([]index.CentroidCardinality, 0, limit*len(i.indexes))
for asr := range asyncResults {
asr = append(asr, rvCentroidCardinalitiesResult...)
if descending {
sort.Slice(asr, func(i, j int) bool {
return asr[i].Cardinality > asr[j].Cardinality
})
} else {
sort.Slice(asr, func(i, j int) bool {
return asr[i].Cardinality < asr[j].Cardinality
})
}
if limit > len(asr) {
limit = len(asr)
}
rvCentroidCardinalitiesResult = asr[:limit]
rvCentroidCardinalities = append(rvCentroidCardinalities, asr...)
}
return rvCentroidCardinalitiesResult, nil
sort.Slice(rvCentroidCardinalities, func(i, j int) bool {
if descending {
return rvCentroidCardinalities[i].Cardinality > rvCentroidCardinalities[j].Cardinality
} else {
return rvCentroidCardinalities[i].Cardinality < rvCentroidCardinalities[j].Cardinality
}
})
if limit > len(rvCentroidCardinalities) {
limit = len(rvCentroidCardinalities)
}
return rvCentroidCardinalities[:limit], nil
}

View File

@@ -20,6 +20,7 @@ import (
"io"
"os"
"path/filepath"
"regexp"
"strconv"
"sync"
"sync/atomic"
@@ -859,6 +860,26 @@ func (i *indexImpl) SearchInContext(ctx context.Context, req *SearchRequest) (sr
} else {
// build terms facet
facetBuilder := facet.NewTermsFacetBuilder(facetRequest.Field, facetRequest.Size)
// Set prefix filter if provided
if facetRequest.TermPrefix != "" {
facetBuilder.SetPrefixFilter(facetRequest.TermPrefix)
}
// Set regex filter if provided
if facetRequest.TermPattern != "" {
// Use cached compiled pattern if available, otherwise compile it now
if facetRequest.compiledPattern != nil {
facetBuilder.SetRegexFilter(facetRequest.compiledPattern)
} else {
regex, err := regexp.Compile(facetRequest.TermPattern)
if err != nil {
return nil, fmt.Errorf("error compiling regex pattern for facet '%s': %v", facetName, err)
}
facetBuilder.SetRegexFilter(regex)
}
}
facetsBuilder.Add(facetName, facetBuilder)
}
}
@@ -1304,6 +1325,9 @@ func (f *indexImplFieldDict) Cardinality() int {
// helper function to remove duplicate entries from slice of strings
func deDuplicate(fields []string) []string {
if len(fields) == 0 {
return fields
}
entries := make(map[string]struct{})
ret := []string{}
for _, entry := range fields {

View File

@@ -92,7 +92,7 @@ func DeletedFields(ori, upd *mapping.IndexMappingImpl) (map[string]*index.Update
// Compare both the mappings based on the document paths
// and create a list of index, docvalues, store differences
// for every single field possible
fieldInfo := make(map[string]*index.UpdateFieldInfo)
fieldInfo := make(map[string]*index.UpdateFieldInfo, len(oriPaths))
for path, info := range oriPaths {
err = addFieldInfo(fieldInfo, info, updPaths[path])
if err != nil {
@@ -109,13 +109,13 @@ func DeletedFields(ori, upd *mapping.IndexMappingImpl) (map[string]*index.Update
// A field cannot be completely deleted with any dynamic value turned on
if info.Deleted {
if upd.IndexDynamic {
return nil, fmt.Errorf("Mapping cannot be removed when index dynamic is true")
return nil, fmt.Errorf("mapping cannot be removed when index dynamic is true")
}
if upd.StoreDynamic {
return nil, fmt.Errorf("Mapping cannot be removed when store dynamic is true")
return nil, fmt.Errorf("mapping cannot be removed when store dynamic is true")
}
if upd.DocValuesDynamic {
return nil, fmt.Errorf("Mapping cannot be removed when docvalues dynamic is true")
return nil, fmt.Errorf("mapping cannot be removed when docvalues dynamic is true")
}
}
}
@@ -191,14 +191,14 @@ func checkUpdatedMapping(ori, upd *mapping.DocumentMapping) error {
// Simple checks to ensure no new field mappings present
// in updated
// Create a map of original field names for O(1) lookup
oriFieldNames := make(map[string]bool, len(ori.Fields))
for _, fMapping := range ori.Fields {
oriFieldNames[fMapping.Name] = true
}
for _, updFMapping := range upd.Fields {
var oriFMapping *mapping.FieldMapping
for _, fMapping := range ori.Fields {
if updFMapping.Name == fMapping.Name {
oriFMapping = fMapping
}
}
if oriFMapping == nil {
if !oriFieldNames[updFMapping.Name] {
return fmt.Errorf("updated index mapping contains new fields")
}
}
@@ -238,10 +238,8 @@ func addPathInfo(paths map[string]*pathInfo, name string, mp *mapping.DocumentMa
// Recursively add path information for all child mappings
for cName, cMapping := range mp.Properties {
var pathName string
if name == "" {
pathName = cName
} else {
pathName := cName
if name != "" {
pathName = name + "." + cName
}
addPathInfo(paths, pathName, cMapping, im, pInfo, rootName)
@@ -460,9 +458,6 @@ func addFieldInfo(fInfo map[string]*index.UpdateFieldInfo, ori, upd *pathInfo) e
}
}
}
if err != nil {
return err
}
return nil
}
@@ -567,19 +562,18 @@ func compareFieldMapping(original, updated *mapping.FieldMapping) (*index.Update
// In such a situation, any conflicting changes found will abort the update process
func validateFieldInfo(newInfo *index.UpdateFieldInfo, fInfo map[string]*index.UpdateFieldInfo,
ori *pathInfo, oriFMapInfo *fieldMapInfo) error {
// Determine field name
fieldName := oriFMapInfo.fieldMapping.Name
if fieldName == "" {
fieldName = oriFMapInfo.parent.path
}
// Construct full name with parent path
var name string
if oriFMapInfo.parent.parentPath == "" {
if oriFMapInfo.fieldMapping.Name == "" {
name = oriFMapInfo.parent.path
} else {
name = oriFMapInfo.fieldMapping.Name
}
name = fieldName
} else {
if oriFMapInfo.fieldMapping.Name == "" {
name = oriFMapInfo.parent.parentPath + "." + oriFMapInfo.parent.path
} else {
name = oriFMapInfo.parent.parentPath + "." + oriFMapInfo.fieldMapping.Name
}
name = oriFMapInfo.parent.parentPath + "." + fieldName
}
if (newInfo.Deleted || newInfo.Index || newInfo.DocValues || newInfo.Store) && ori.dynamic {
return fmt.Errorf("updated field is under a dynamic property")

View File

@@ -52,7 +52,7 @@ type DocumentMapping struct {
}
func (dm *DocumentMapping) Validate(cache *registry.Cache,
parentName string, fieldAliasCtx map[string]*FieldMapping,
path []string, fieldAliasCtx map[string]*FieldMapping,
) error {
var err error
if dm.DefaultAnalyzer != "" {
@@ -68,11 +68,7 @@ func (dm *DocumentMapping) Validate(cache *registry.Cache,
}
}
for propertyName, property := range dm.Properties {
newParent := propertyName
if parentName != "" {
newParent = fmt.Sprintf("%s.%s", parentName, propertyName)
}
err = property.Validate(cache, newParent, fieldAliasCtx)
err = property.Validate(cache, append(path, propertyName), fieldAliasCtx)
if err != nil {
return err
}
@@ -96,7 +92,7 @@ func (dm *DocumentMapping) Validate(cache *registry.Cache,
return err
}
}
err := validateFieldMapping(field, parentName, fieldAliasCtx)
err := validateFieldMapping(field, path, fieldAliasCtx)
if err != nil {
return err
}

View File

@@ -191,13 +191,16 @@ func (im *IndexMappingImpl) Validate() error {
return err
}
}
// fieldAliasCtx is used to detect any field alias conflicts across the entire mapping
// the map will hold the fully qualified field name to FieldMapping, so we can
// check for conflicts as we validate each DocumentMapping.
fieldAliasCtx := make(map[string]*FieldMapping)
err = im.DefaultMapping.Validate(im.cache, "", fieldAliasCtx)
err = im.DefaultMapping.Validate(im.cache, []string{}, fieldAliasCtx)
if err != nil {
return err
}
for _, docMapping := range im.TypeMapping {
err = docMapping.Validate(im.cache, "", fieldAliasCtx)
err = docMapping.Validate(im.cache, []string{}, fieldAliasCtx)
if err != nil {
return err
}

View File

@@ -38,7 +38,7 @@ func (fm *FieldMapping) processVectorBase64(propertyMightBeVector interface{},
// -----------------------------------------------------------------------------
// document validation functions
func validateFieldMapping(field *FieldMapping, parentName string,
func validateFieldMapping(field *FieldMapping, path []string,
fieldAliasCtx map[string]*FieldMapping) error {
return validateFieldType(field)
}

View File

@@ -20,6 +20,7 @@ package mapping
import (
"fmt"
"reflect"
"slices"
"github.com/blevesearch/bleve/v2/document"
"github.com/blevesearch/bleve/v2/util"
@@ -141,15 +142,27 @@ func (fm *FieldMapping) processVector(propertyMightBeVector interface{},
if !ok {
return false
}
// Apply defaults for similarity and optimization if not set
similarity := fm.Similarity
if similarity == "" {
similarity = index.DefaultVectorSimilarityMetric
}
vectorIndexOptimizedFor := fm.VectorIndexOptimizedFor
if vectorIndexOptimizedFor == "" {
vectorIndexOptimizedFor = index.DefaultIndexOptimization
}
// normalize raw vector if similarity is cosine
if fm.Similarity == index.CosineSimilarity {
vector = NormalizeVector(vector)
// Since the vector can be multi-vector (flattened array of multiple vectors),
// we use NormalizeMultiVector to normalize each sub-vector independently.
if similarity == index.CosineSimilarity {
vector = NormalizeMultiVector(vector, fm.Dims)
}
fieldName := getFieldName(pathString, path, fm)
options := fm.Options()
field := document.NewVectorFieldWithIndexingOptions(fieldName, indexes, vector,
fm.Dims, fm.Similarity, fm.VectorIndexOptimizedFor, options)
fm.Dims, similarity, vectorIndexOptimizedFor, options)
context.doc.AddField(field)
// "_all" composite field is not applicable for vector field
@@ -163,20 +176,29 @@ func (fm *FieldMapping) processVectorBase64(propertyMightBeVectorBase64 interfac
if !ok {
return
}
// Apply defaults for similarity and optimization if not set
similarity := fm.Similarity
if similarity == "" {
similarity = index.DefaultVectorSimilarityMetric
}
vectorIndexOptimizedFor := fm.VectorIndexOptimizedFor
if vectorIndexOptimizedFor == "" {
vectorIndexOptimizedFor = index.DefaultIndexOptimization
}
decodedVector, err := document.DecodeVector(encodedString)
if err != nil || len(decodedVector) != fm.Dims {
return
}
// normalize raw vector if similarity is cosine
if fm.Similarity == index.CosineSimilarity {
// normalize raw vector if similarity is cosine, multi-vector is not supported
// for base64 encoded vectors, so we use NormalizeVector directly.
if similarity == index.CosineSimilarity {
decodedVector = NormalizeVector(decodedVector)
}
fieldName := getFieldName(pathString, path, fm)
options := fm.Options()
field := document.NewVectorFieldWithIndexingOptions(fieldName, indexes, decodedVector,
fm.Dims, fm.Similarity, fm.VectorIndexOptimizedFor, options)
fm.Dims, similarity, vectorIndexOptimizedFor, options)
context.doc.AddField(field)
// "_all" composite field is not applicable for vector_base64 field
@@ -186,87 +208,121 @@ func (fm *FieldMapping) processVectorBase64(propertyMightBeVectorBase64 interfac
// -----------------------------------------------------------------------------
// document validation functions
func validateFieldMapping(field *FieldMapping, parentName string,
func validateFieldMapping(field *FieldMapping, path []string,
fieldAliasCtx map[string]*FieldMapping) error {
switch field.Type {
case "vector", "vector_base64":
return validateVectorFieldAlias(field, parentName, fieldAliasCtx)
return validateVectorFieldAlias(field, path, fieldAliasCtx)
default: // non-vector field
return validateFieldType(field)
}
}
func validateVectorFieldAlias(field *FieldMapping, parentName string,
func validateVectorFieldAlias(field *FieldMapping, path []string,
fieldAliasCtx map[string]*FieldMapping) error {
if field.Name == "" {
field.Name = parentName
// fully qualified field name
pathString := encodePath(path)
// check if field has a name set, else use path to compute effective name
effectiveFieldName := getFieldName(pathString, path, field)
// Compute effective values for validation
effectiveSimilarity := field.Similarity
if effectiveSimilarity == "" {
effectiveSimilarity = index.DefaultVectorSimilarityMetric
}
effectiveOptimizedFor := field.VectorIndexOptimizedFor
if effectiveOptimizedFor == "" {
effectiveOptimizedFor = index.DefaultIndexOptimization
}
if field.Similarity == "" {
field.Similarity = index.DefaultVectorSimilarityMetric
}
if field.VectorIndexOptimizedFor == "" {
field.VectorIndexOptimizedFor = index.DefaultIndexOptimization
}
if _, exists := index.SupportedVectorIndexOptimizations[field.VectorIndexOptimizedFor]; !exists {
// if an unsupported config is provided, override to default
field.VectorIndexOptimizedFor = index.DefaultIndexOptimization
}
// following fields are not applicable for vector
// thus, we set them to default values
field.IncludeInAll = false
field.IncludeTermVectors = false
field.Store = false
field.DocValues = false
field.SkipFreqNorm = true
// # If alias is present, validate the field options as per the alias
// # If alias is present, validate the field options as per the alias.
// note: reading from a nil map is safe
if fieldAlias, ok := fieldAliasCtx[field.Name]; ok {
if fieldAlias, ok := fieldAliasCtx[effectiveFieldName]; ok {
if field.Dims != fieldAlias.Dims {
return fmt.Errorf("field: '%s', invalid alias "+
"(different dimensions %d and %d)", fieldAlias.Name, field.Dims,
"(different dimensions %d and %d)", effectiveFieldName, field.Dims,
fieldAlias.Dims)
}
if field.Similarity != fieldAlias.Similarity {
// Compare effective similarity values
aliasSimilarity := fieldAlias.Similarity
if aliasSimilarity == "" {
aliasSimilarity = index.DefaultVectorSimilarityMetric
}
if effectiveSimilarity != aliasSimilarity {
return fmt.Errorf("field: '%s', invalid alias "+
"(different similarity values %s and %s)", fieldAlias.Name,
field.Similarity, fieldAlias.Similarity)
"(different similarity values %s and %s)", effectiveFieldName,
effectiveSimilarity, aliasSimilarity)
}
// Compare effective vector index optimization values
aliasOptimizedFor := fieldAlias.VectorIndexOptimizedFor
if aliasOptimizedFor == "" {
aliasOptimizedFor = index.DefaultIndexOptimization
}
if effectiveOptimizedFor != aliasOptimizedFor {
return fmt.Errorf("field: '%s', invalid alias "+
"(different vector index optimization values %s and %s)", effectiveFieldName,
effectiveOptimizedFor, aliasOptimizedFor)
}
return nil
}
// # Validate field options
// Vector dimensions must be within allowed range
if field.Dims < MinVectorDims || field.Dims > MaxVectorDims {
return fmt.Errorf("field: '%s', invalid vector dimension: %d,"+
" value should be in range (%d, %d)", field.Name, field.Dims,
" value should be in range [%d, %d]", effectiveFieldName, field.Dims,
MinVectorDims, MaxVectorDims)
}
if _, ok := index.SupportedVectorSimilarityMetrics[field.Similarity]; !ok {
// Similarity metric must be supported
if _, ok := index.SupportedVectorSimilarityMetrics[effectiveSimilarity]; !ok {
return fmt.Errorf("field: '%s', invalid similarity "+
"metric: '%s', valid metrics are: %+v", field.Name, field.Similarity,
"metric: '%s', valid metrics are: %+v", effectiveFieldName, effectiveSimilarity,
reflect.ValueOf(index.SupportedVectorSimilarityMetrics).MapKeys())
}
// Vector index optimization must be supported
if _, ok := index.SupportedVectorIndexOptimizations[effectiveOptimizedFor]; !ok {
return fmt.Errorf("field: '%s', invalid vector index "+
"optimization: '%s', valid optimizations are: %+v", effectiveFieldName,
effectiveOptimizedFor,
reflect.ValueOf(index.SupportedVectorIndexOptimizations).MapKeys())
}
if fieldAliasCtx != nil { // writing to a nil map is unsafe
fieldAliasCtx[field.Name] = field
fieldAliasCtx[effectiveFieldName] = field
}
return nil
}
// NormalizeVector normalizes a single vector to unit length.
// It makes a copy of the input vector to avoid modifying it in-place.
func NormalizeVector(vec []float32) []float32 {
// make a copy of the vector to avoid modifying the original
// vector in-place
vecCopy := make([]float32, len(vec))
copy(vecCopy, vec)
vecCopy := slices.Clone(vec)
// normalize the vector copy using in-place normalization provided by faiss
return faiss.NormalizeVector(vecCopy)
}
// NormalizeMultiVector normalizes each sub-vector of size `dims` independently.
// For a flattened array containing multiple vectors, each sub-vector is
// normalized separately to unit length.
// It makes a copy of the input vector to avoid modifying it in-place.
func NormalizeMultiVector(vec []float32, dims int) []float32 {
if len(vec) == 0 || dims <= 0 || len(vec)%dims != 0 {
return vec
}
// Single vector - delegate to NormalizeVector
if len(vec) == dims {
return NormalizeVector(vec)
}
// Multi-vector - make a copy to avoid modifying the original
result := slices.Clone(vec)
// Normalize each sub-vector in-place
for i := 0; i < len(result); i += dims {
faiss.NormalizeVector(result[i : i+dims])
}
return result
}

View File

@@ -99,7 +99,7 @@ func (r *rescorer) rescore(ftsHits, knnHits search.DocumentMatchCollection) (sea
switch r.req.Score {
case ScoreRRF:
res := fusion.ReciprocalRankFusion(
fusionResult = fusion.ReciprocalRankFusion(
mergedHits,
r.origBoosts,
r.req.Params.ScoreRankConstant,
@@ -107,16 +107,14 @@ func (r *rescorer) rescore(ftsHits, knnHits search.DocumentMatchCollection) (sea
numKNNQueries(r.req),
r.req.Explain,
)
fusionResult = &res
case ScoreRSF:
res := fusion.RelativeScoreFusion(
fusionResult = fusion.RelativeScoreFusion(
mergedHits,
r.origBoosts,
r.req.Params.ScoreWindowSize,
numKNNQueries(r.req),
r.req.Explain,
)
fusionResult = &res
}
return fusionResult.Hits, fusionResult.Total, fusionResult.MaxScore

View File

@@ -17,8 +17,10 @@ package bleve
import (
"fmt"
"reflect"
"regexp"
"sort"
"strconv"
"strings"
"time"
"github.com/blevesearch/bleve/v2/analysis"
@@ -147,8 +149,13 @@ type numericRange struct {
type FacetRequest struct {
Size int `json:"size"`
Field string `json:"field"`
TermPrefix string `json:"term_prefix,omitempty"`
TermPattern string `json:"term_pattern,omitempty"`
NumericRanges []*numericRange `json:"numeric_ranges,omitempty"`
DateTimeRanges []*dateTimeRange `json:"date_ranges,omitempty"`
// Compiled regex pattern (cached during validation)
compiledPattern *regexp.Regexp
}
// NewFacetRequest creates a facet on the specified
@@ -161,7 +168,26 @@ func NewFacetRequest(field string, size int) *FacetRequest {
}
}
// SetPrefixFilter sets the prefix filter for term facets.
func (fr *FacetRequest) SetPrefixFilter(prefix string) {
fr.TermPrefix = prefix
}
// SetRegexFilter sets the regex pattern filter for term facets.
func (fr *FacetRequest) SetRegexFilter(pattern string) {
fr.TermPattern = pattern
}
func (fr *FacetRequest) Validate() error {
// Validate regex pattern if provided and cache the compiled regex
if fr.TermPattern != "" {
compiled, err := regexp.Compile(fr.TermPattern)
if err != nil {
return fmt.Errorf("invalid term pattern: %v", err)
}
fr.compiledPattern = compiled
}
nrCount := len(fr.NumericRanges)
drCount := len(fr.DateTimeRanges)
if nrCount > 0 && drCount > 0 {
@@ -546,49 +572,74 @@ func (sr *SearchResult) Size() int {
}
func (sr *SearchResult) String() string {
rv := ""
rv := &strings.Builder{}
if sr.Total > 0 {
if sr.Request != nil && sr.Request.Size > 0 {
rv = fmt.Sprintf("%d matches, showing %d through %d, took %s\n", sr.Total, sr.Request.From+1, sr.Request.From+len(sr.Hits), sr.Took)
switch {
case sr.Request != nil && sr.Request.Size > 0:
start := sr.Request.From + 1
end := sr.Request.From + len(sr.Hits)
fmt.Fprintf(rv, "%d matches, showing %d through %d, took %s\n", sr.Total, start, end, sr.Took)
for i, hit := range sr.Hits {
rv += fmt.Sprintf("%5d. %s (%f)\n", i+sr.Request.From+1, hit.ID, hit.Score)
for fragmentField, fragments := range hit.Fragments {
rv += fmt.Sprintf("\t%s\n", fragmentField)
for _, fragment := range fragments {
rv += fmt.Sprintf("\t\t%s\n", fragment)
}
}
for otherFieldName, otherFieldValue := range hit.Fields {
if _, ok := hit.Fragments[otherFieldName]; !ok {
rv += fmt.Sprintf("\t%s\n", otherFieldName)
rv += fmt.Sprintf("\t\t%v\n", otherFieldValue)
}
}
rv = formatHit(rv, hit, start+i)
}
} else {
rv = fmt.Sprintf("%d matches, took %s\n", sr.Total, sr.Took)
case sr.Request == nil:
fmt.Fprintf(rv, "%d matches, took %s\n", sr.Total, sr.Took)
for i, hit := range sr.Hits {
rv = formatHit(rv, hit, i+1)
}
default:
fmt.Fprintf(rv, "%d matches, took %s\n", sr.Total, sr.Took)
}
} else {
rv = "No matches"
fmt.Fprintf(rv, "No matches\n")
}
if len(sr.Facets) > 0 {
rv += "Facets:\n"
fmt.Fprintf(rv, "Facets:\n")
for fn, f := range sr.Facets {
rv += fmt.Sprintf("%s(%d)\n", fn, f.Total)
fmt.Fprintf(rv, "%s(%d)\n", fn, f.Total)
for _, t := range f.Terms.Terms() {
rv += fmt.Sprintf("\t%s(%d)\n", t.Term, t.Count)
fmt.Fprintf(rv, "\t%s(%d)\n", t.Term, t.Count)
}
for _, n := range f.NumericRanges {
rv += fmt.Sprintf("\t%s(%d)\n", n.Name, n.Count)
fmt.Fprintf(rv, "\t%s(%d)\n", n.Name, n.Count)
}
for _, d := range f.DateRanges {
rv += fmt.Sprintf("\t%s(%d)\n", d.Name, d.Count)
fmt.Fprintf(rv, "\t%s(%d)\n", d.Name, d.Count)
}
if f.Other != 0 {
rv += fmt.Sprintf("\tOther(%d)\n", f.Other)
fmt.Fprintf(rv, "\tOther(%d)\n", f.Other)
}
}
}
return rv.String()
}
// formatHit is a helper function to format a single hit in the search result for
// the String() method of SearchResult
func formatHit(rv *strings.Builder, hit *search.DocumentMatch, hitNumber int) *strings.Builder {
fmt.Fprintf(rv, "%5d. %s (%f)\n", hitNumber, hit.ID, hit.Score)
for fragmentField, fragments := range hit.Fragments {
fmt.Fprintf(rv, "\t%s\n", fragmentField)
for _, fragment := range fragments {
fmt.Fprintf(rv, "\t\t%s\n", fragment)
}
}
for otherFieldName, otherFieldValue := range hit.Fields {
if _, ok := hit.Fragments[otherFieldName]; !ok {
fmt.Fprintf(rv, "\t%s\n", otherFieldName)
fmt.Fprintf(rv, "\t\t%v\n", otherFieldValue)
}
}
if len(hit.DecodedSort) > 0 {
fmt.Fprintf(rv, "\t_sort: [")
for k, v := range hit.DecodedSort {
if k > 0 {
fmt.Fprintf(rv, ", ")
}
fmt.Fprintf(rv, "%v", v)
}
fmt.Fprintf(rv, "]\n")
}
return rv
}

View File

@@ -15,7 +15,9 @@
package facet
import (
"bytes"
"reflect"
"regexp"
"sort"
"github.com/blevesearch/bleve/v2/search"
@@ -30,12 +32,14 @@ func init() {
}
type TermsFacetBuilder struct {
size int
field string
termsCount map[string]int
total int
missing int
sawValue bool
size int
field string
prefixBytes []byte
regex *regexp.Regexp
termsCount map[string]int
total int
missing int
sawValue bool
}
func NewTermsFacetBuilder(field string, size int) *TermsFacetBuilder {
@@ -48,7 +52,16 @@ func NewTermsFacetBuilder(field string, size int) *TermsFacetBuilder {
func (fb *TermsFacetBuilder) Size() int {
sizeInBytes := reflectStaticSizeTermsFacetBuilder + size.SizeOfPtr +
len(fb.field)
len(fb.field) +
len(fb.prefixBytes) +
size.SizeOfPtr // regex pointer (does not include actual regexp.Regexp object size)
// Estimate regex object size if present.
if fb.regex != nil {
// This is only the static size of regexp.Regexp struct, not including heap allocations.
sizeInBytes += int(reflect.TypeOf(*fb.regex).Size())
// NOTE: Actual memory usage of regexp.Regexp may be higher due to internal allocations.
}
for k := range fb.termsCount {
sizeInBytes += size.SizeOfString + len(k) +
@@ -62,10 +75,39 @@ func (fb *TermsFacetBuilder) Field() string {
return fb.field
}
// SetPrefixFilter sets the prefix filter for term facets.
func (fb *TermsFacetBuilder) SetPrefixFilter(prefix string) {
if prefix != "" {
fb.prefixBytes = []byte(prefix)
} else {
fb.prefixBytes = nil
}
}
// SetRegexFilter sets the compiled regex filter for term facets.
func (fb *TermsFacetBuilder) SetRegexFilter(regex *regexp.Regexp) {
fb.regex = regex
}
func (fb *TermsFacetBuilder) UpdateVisitor(term []byte) {
fb.sawValue = true
fb.termsCount[string(term)] = fb.termsCount[string(term)] + 1
// Total represents all terms visited, not just matching ones.
// This is necessary for the "Other" calculation.
fb.total++
// Fast prefix check on []byte - zero allocation
if len(fb.prefixBytes) > 0 && !bytes.HasPrefix(term, fb.prefixBytes) {
return
}
// Fast regex check on []byte - zero allocation
if fb.regex != nil && !fb.regex.Match(term) {
return
}
// Only convert to string if term matches filters
termStr := string(term)
fb.sawValue = true
fb.termsCount[termStr] = fb.termsCount[termStr] + 1
}
func (fb *TermsFacetBuilder) StartDoc() {

View File

@@ -15,7 +15,6 @@
package query
import (
"bytes"
"context"
"encoding/json"
"fmt"
@@ -203,7 +202,7 @@ func (q *BooleanQuery) Searcher(ctx context.Context, i index.IndexReader, m mapp
return false
}
// Compare document IDs
cmp := bytes.Compare(refDoc.IndexInternalID, d.IndexInternalID)
cmp := refDoc.IndexInternalID.Compare(d.IndexInternalID)
if cmp < 0 {
// filterSearcher is behind the current document, Advance() it
refDoc, err = filterSearcher.Advance(sctx, d.IndexInternalID)
@@ -211,7 +210,7 @@ func (q *BooleanQuery) Searcher(ctx context.Context, i index.IndexReader, m mapp
return false
}
// After advance, check if they're now equal
return bytes.Equal(refDoc.IndexInternalID, d.IndexInternalID)
cmp = refDoc.IndexInternalID.Compare(d.IndexInternalID)
}
// cmp >= 0: either equal (match) or filterSearcher is ahead (no match)
return cmp == 0

View File

@@ -53,7 +53,7 @@ func (q *KNNQuery) SetK(k int64) {
q.K = k
}
func (q *KNNQuery) SetFieldVal(field string) {
func (q *KNNQuery) SetField(field string) {
q.VectorField = field
}

View File

@@ -88,7 +88,10 @@ func (s *DisjunctionQueryScorer) Score(ctx *search.SearchContext, constituents [
func (s *DisjunctionQueryScorer) ScoreAndExplBreakdown(ctx *search.SearchContext, constituents []*search.DocumentMatch,
matchingIdxs []int, originalPositions []int, countTotal int) *search.DocumentMatch {
scoreBreakdown := make(map[int]float64)
rv := constituents[0]
if rv.ScoreBreakdown == nil {
rv.ScoreBreakdown = make(map[int]float64, len(constituents))
}
var childrenExplanations []*search.Explanation
if s.options.Explain {
// since we want to notify which expl belongs to which matched searcher within the disjunction searcher
@@ -104,7 +107,7 @@ func (s *DisjunctionQueryScorer) ScoreAndExplBreakdown(ctx *search.SearchContext
// scorer used in disjunction heap searcher
index = matchingIdxs[i]
}
scoreBreakdown[index] = docMatch.Score
rv.ScoreBreakdown[index] = docMatch.Score
if s.options.Explain {
childrenExplanations[index] = docMatch.Expl
}
@@ -113,9 +116,6 @@ func (s *DisjunctionQueryScorer) ScoreAndExplBreakdown(ctx *search.SearchContext
if s.options.Explain {
explBreakdown = &search.Explanation{Children: childrenExplanations}
}
rv := constituents[0]
rv.ScoreBreakdown = scoreBreakdown
rv.Expl = explBreakdown
rv.FieldTermLocations = search.MergeFieldTermLocations(
rv.FieldTermLocations, constituents[1:])

View File

@@ -207,20 +207,29 @@ func (dm *DocumentMatch) Reset() *DocumentMatch {
indexInternalID := dm.IndexInternalID
// remember the []interface{} used for sort
sort := dm.Sort
// remember the []string used for decoded sort
decodedSort := dm.DecodedSort
// remember the FieldTermLocations backing array
ftls := dm.FieldTermLocations
for i := range ftls { // recycle the ArrayPositions of each location
ftls[i].Location.ArrayPositions = ftls[i].Location.ArrayPositions[:0]
}
// remember the score breakdown map
scoreBreakdown := dm.ScoreBreakdown
// clear out the score breakdown map
clear(scoreBreakdown)
// idiom to copy over from empty DocumentMatch (0 allocations)
*dm = DocumentMatch{}
// reuse the []byte already allocated (and reset len to 0)
dm.IndexInternalID = indexInternalID[:0]
// reuse the []interface{} already allocated (and reset len to 0)
dm.Sort = sort[:0]
dm.DecodedSort = dm.DecodedSort[:0]
// reuse the []string already allocated (and reset len to 0)
dm.DecodedSort = decodedSort[:0]
// reuse the FieldTermLocations already allocated (and reset len to 0)
dm.FieldTermLocations = ftls[:0]
// reuse the score breakdown map already allocated (after clearing it)
dm.ScoreBreakdown = scoreBreakdown
return dm
}

View File

@@ -84,7 +84,7 @@ func (s *KNNSearcher) VectorOptimize(ctx context.Context, octx index.VectorOptim
func (s *KNNSearcher) Advance(ctx *search.SearchContext, ID index.IndexInternalID) (
*search.DocumentMatch, error) {
knnMatch, err := s.vectorReader.Next(s.vd.Reset())
knnMatch, err := s.vectorReader.Advance(ID, s.vd.Reset())
if err != nil {
return nil, err
}

View File

@@ -288,10 +288,15 @@ func createKNNQuery(req *SearchRequest, knnFilterResults map[int]index.EligibleD
// If it's a filtered kNN but has no eligible filter hits, then
// do not run the kNN query.
if selector, exists := knnFilterResults[i]; exists && selector == nil {
// if the kNN query is filtered and has no eligible filter hits, then
// do not run the kNN query, so we add a match_none query to the subQueries.
// this will ensure that the score breakdown is set to 0 for this kNN query.
subQueries = append(subQueries, NewMatchNoneQuery())
kArray = append(kArray, 0)
continue
}
knnQuery := query.NewKNNQuery(knn.Vector)
knnQuery.SetFieldVal(knn.Field)
knnQuery.SetField(knn.Field)
knnQuery.SetK(knn.K)
knnQuery.SetBoost(knn.Boost.Value())
knnQuery.SetParams(knn.Params)
@@ -381,7 +386,7 @@ func addSortAndFieldsToKNNHits(req *SearchRequest, knnHits []*search.DocumentMat
return nil
}
func (i *indexImpl) runKnnCollector(ctx context.Context, req *SearchRequest, reader index.IndexReader, preSearch bool) ([]*search.DocumentMatch, error) {
func (i *indexImpl) runKnnCollector(ctx context.Context, req *SearchRequest, reader index.IndexReader, preSearch bool) (knnHits []*search.DocumentMatch, err error) {
// Maps the index of a KNN query in the request to its pre-filter result:
// - If the KNN query is **not filtered**, the value will be `nil`.
// - If the KNN query **is filtered**, the value will be an eligible document selector
@@ -401,21 +406,33 @@ func (i *indexImpl) runKnnCollector(ctx context.Context, req *SearchRequest, rea
continue
}
// Applies to all supported types of queries.
filterSearcher, _ := filterQ.Searcher(ctx, reader, i.m, search.SearcherOptions{
filterSearcher, err := filterQ.Searcher(ctx, reader, i.m, search.SearcherOptions{
Score: "none", // just want eligible hits --> don't compute scores if not needed
})
if err != nil {
return nil, err
}
// Using the index doc count to determine collector size since we do not
// have an estimate of the number of eligible docs in the index yet.
indexDocCount, err := i.DocCount()
if err != nil {
// close the searcher before returning
filterSearcher.Close()
return nil, err
}
filterColl := collector.NewEligibleCollector(int(indexDocCount))
err = filterColl.Collect(ctx, filterSearcher, reader)
if err != nil {
// close the searcher before returning
filterSearcher.Close()
return nil, err
}
knnFilterResults[idx] = filterColl.EligibleSelector()
// Close the filter searcher, as we are done with it.
err = filterSearcher.Close()
if err != nil {
return nil, err
}
}
// Add the filter hits when creating the kNN query
@@ -429,12 +446,17 @@ func (i *indexImpl) runKnnCollector(ctx context.Context, req *SearchRequest, rea
if err != nil {
return nil, err
}
defer func() {
if serr := knnSearcher.Close(); err == nil && serr != nil {
err = serr
}
}()
knnCollector := collector.NewKNNCollector(kArray, sumOfK)
err = knnCollector.Collect(ctx, knnSearcher, reader)
if err != nil {
return nil, err
}
knnHits := knnCollector.Results()
knnHits = knnCollector.Results()
if !preSearch {
knnHits = finalizeKNNResults(req, knnHits)
}

View File

@@ -19,15 +19,11 @@ package zap
import (
"encoding/binary"
"encoding/json"
"math"
"reflect"
"github.com/RoaringBitmap/roaring/v2"
"github.com/RoaringBitmap/roaring/v2/roaring64"
"github.com/bits-and-blooms/bitset"
index "github.com/blevesearch/bleve_index_api"
faiss "github.com/blevesearch/go-faiss"
segment "github.com/blevesearch/scorch_segment_api/v2"
)
@@ -272,45 +268,7 @@ func (vpItr *VecPostingsIterator) BytesWritten() uint64 {
return 0
}
// vectorIndexWrapper conforms to scorch_segment_api's VectorIndex interface
type vectorIndexWrapper struct {
search func(qVector []float32, k int64,
params json.RawMessage) (segment.VecPostingsList, error)
searchWithFilter func(qVector []float32, k int64, eligibleDocIDs []uint64,
params json.RawMessage) (segment.VecPostingsList, error)
close func()
size func() uint64
obtainKCentroidCardinalitiesFromIVFIndex func(limit int, descending bool) (
[]index.CentroidCardinality, error)
}
func (i *vectorIndexWrapper) Search(qVector []float32, k int64,
params json.RawMessage) (
segment.VecPostingsList, error) {
return i.search(qVector, k, params)
}
func (i *vectorIndexWrapper) SearchWithFilter(qVector []float32, k int64,
eligibleDocIDs []uint64, params json.RawMessage) (
segment.VecPostingsList, error) {
return i.searchWithFilter(qVector, k, eligibleDocIDs, params)
}
func (i *vectorIndexWrapper) Close() {
i.close()
}
func (i *vectorIndexWrapper) Size() uint64 {
return i.size()
}
func (i *vectorIndexWrapper) ObtainKCentroidCardinalitiesFromIVFIndex(limit int, descending bool) (
[]index.CentroidCardinality, error) {
return i.obtainKCentroidCardinalitiesFromIVFIndex(limit, descending)
}
// InterpretVectorIndex returns a construct of closures (vectorIndexWrapper)
// InterpretVectorIndex returns a struct based implementation (vectorIndexWrapper)
// that will allow the caller to -
// (1) search within an attached vector index
// (2) search limited to a subset of documents within an attached vector index
@@ -319,248 +277,18 @@ func (i *vectorIndexWrapper) ObtainKCentroidCardinalitiesFromIVFIndex(limit int,
func (sb *SegmentBase) InterpretVectorIndex(field string, requiresFiltering bool,
except *roaring.Bitmap) (
segment.VectorIndex, error) {
// Params needed for the closures
var vecIndex *faiss.IndexImpl
var vecDocIDMap map[int64]uint32
var docVecIDMap map[uint32][]int64
var vectorIDsToExclude []int64
var fieldIDPlus1 uint16
var vecIndexSize uint64
// Utility function to add the corresponding docID and scores for each vector
// returned after the kNN query to the newly
// created vecPostingsList
addIDsToPostingsList := func(pl *VecPostingsList, ids []int64, scores []float32) {
for i := 0; i < len(ids); i++ {
vecID := ids[i]
// Checking if it's present in the vecDocIDMap.
// If -1 is returned as an ID(insufficient vectors), this will ensure
// it isn't added to the final postings list.
if docID, ok := vecDocIDMap[vecID]; ok {
code := getVectorCode(docID, scores[i])
pl.postings.Add(code)
}
}
}
var (
wrapVecIndex = &vectorIndexWrapper{
search: func(qVector []float32, k int64, params json.RawMessage) (
segment.VecPostingsList, error) {
// 1. returned postings list (of type PostingsList) has two types of information - docNum and its score.
// 2. both the values can be represented using roaring bitmaps.
// 3. the Iterator (of type PostingsIterator) returned would operate in terms of VecPostings.
// 4. VecPostings would just have the docNum and the score. Every call of Next()
// and Advance just returns the next VecPostings. The caller would do a vp.Number()
// and the Score() to get the corresponding values
rv := &VecPostingsList{
except: nil, // todo: handle the except bitmap within postings iterator.
postings: roaring64.New(),
}
if vecIndex == nil || vecIndex.D() != len(qVector) {
// vector index not found or dimensionality mismatched
return rv, nil
}
scores, ids, err := vecIndex.SearchWithoutIDs(qVector, k,
vectorIDsToExclude, params)
if err != nil {
return nil, err
}
addIDsToPostingsList(rv, ids, scores)
return rv, nil
},
searchWithFilter: func(qVector []float32, k int64,
eligibleDocIDs []uint64, params json.RawMessage) (
segment.VecPostingsList, error) {
// 1. returned postings list (of type PostingsList) has two types of information - docNum and its score.
// 2. both the values can be represented using roaring bitmaps.
// 3. the Iterator (of type PostingsIterator) returned would operate in terms of VecPostings.
// 4. VecPostings would just have the docNum and the score. Every call of Next()
// and Advance just returns the next VecPostings. The caller would do a vp.Number()
// and the Score() to get the corresponding values
rv := &VecPostingsList{
except: nil, // todo: handle the except bitmap within postings iterator.
postings: roaring64.New(),
}
if vecIndex == nil || vecIndex.D() != len(qVector) {
// vector index not found or dimensionality mismatched
return rv, nil
}
// Check and proceed only if non-zero documents eligible per the filter query.
if len(eligibleDocIDs) == 0 {
return rv, nil
}
// If every element in the index is eligible (full selectivity),
// then this can basically be considered unfiltered kNN.
if len(eligibleDocIDs) == int(sb.numDocs) {
scores, ids, err := vecIndex.SearchWithoutIDs(qVector, k,
vectorIDsToExclude, params)
if err != nil {
return nil, err
}
addIDsToPostingsList(rv, ids, scores)
return rv, nil
}
// vector IDs corresponding to the local doc numbers to be
// considered for the search
vectorIDsToInclude := make([]int64, 0, len(eligibleDocIDs))
for _, id := range eligibleDocIDs {
vecIDs := docVecIDMap[uint32(id)]
// In the common case where vecIDs has only one element, which occurs
// when a document has only one vector field, we can
// avoid the unnecessary overhead of slice unpacking (append(vecIDs...)).
// Directly append the single element for efficiency.
if len(vecIDs) == 1 {
vectorIDsToInclude = append(vectorIDsToInclude, vecIDs[0])
} else {
vectorIDsToInclude = append(vectorIDsToInclude, vecIDs...)
}
}
// In case a doc has invalid vector fields but valid non-vector fields,
// filter hit IDs may be ineligible for the kNN since the document does
// not have any/valid vectors.
if len(vectorIDsToInclude) == 0 {
return rv, nil
}
// If the index is not an IVF index, then the search can be
// performed directly, using the Flat index.
if !vecIndex.IsIVFIndex() {
// vector IDs corresponding to the local doc numbers to be
// considered for the search
scores, ids, err := vecIndex.SearchWithIDs(qVector, k,
vectorIDsToInclude, params)
if err != nil {
return nil, err
}
addIDsToPostingsList(rv, ids, scores)
return rv, nil
}
// Determining which clusters, identified by centroid ID,
// have at least one eligible vector and hence, ought to be
// probed.
clusterVectorCounts, err := vecIndex.ObtainClusterVectorCountsFromIVFIndex(vectorIDsToInclude)
if err != nil {
return nil, err
}
var selector faiss.Selector
// If there are more elements to be included than excluded, it
// might be quicker to use an exclusion selector as a filter
// instead of an inclusion selector.
if float32(len(eligibleDocIDs))/float32(len(docVecIDMap)) > 0.5 {
// Use a bitset to efficiently track eligible document IDs.
// This reduces the lookup cost when checking if a document ID is eligible,
// compared to using a map or slice.
bs := bitset.New(uint(len(eligibleDocIDs)))
for _, docID := range eligibleDocIDs {
bs.Set(uint(docID))
}
ineligibleVectorIDs := make([]int64, 0, len(vecDocIDMap)-len(vectorIDsToInclude))
for docID, vecIDs := range docVecIDMap {
// Check if the document ID is NOT in the eligible set, marking it as ineligible.
if !bs.Test(uint(docID)) {
// In the common case where vecIDs has only one element, which occurs
// when a document has only one vector field, we can
// avoid the unnecessary overhead of slice unpacking (append(vecIDs...)).
// Directly append the single element for efficiency.
if len(vecIDs) == 1 {
ineligibleVectorIDs = append(ineligibleVectorIDs, vecIDs[0])
} else {
ineligibleVectorIDs = append(ineligibleVectorIDs, vecIDs...)
}
}
}
selector, err = faiss.NewIDSelectorNot(ineligibleVectorIDs)
} else {
selector, err = faiss.NewIDSelectorBatch(vectorIDsToInclude)
}
if err != nil {
return nil, err
}
// If no error occurred during the creation of the selector, then
// it should be deleted once the search is complete.
defer selector.Delete()
// Ordering the retrieved centroid IDs by increasing order
// of distance i.e. decreasing order of proximity to query vector.
centroidIDs := make([]int64, 0, len(clusterVectorCounts))
for centroidID := range clusterVectorCounts {
centroidIDs = append(centroidIDs, centroidID)
}
closestCentroidIDs, centroidDistances, err :=
vecIndex.ObtainClustersWithDistancesFromIVFIndex(qVector, centroidIDs)
if err != nil {
return nil, err
}
// Getting the nprobe value set at index time.
nprobe := int(vecIndex.GetNProbe())
// Determining the minimum number of centroids to be probed
// to ensure that at least 'k' vectors are collected while
// examining at least 'nprobe' centroids.
var eligibleDocsTillNow int64
minEligibleCentroids := len(closestCentroidIDs)
for i, centroidID := range closestCentroidIDs {
eligibleDocsTillNow += clusterVectorCounts[centroidID]
// Stop once we've examined at least 'nprobe' centroids and
// collected at least 'k' vectors.
if eligibleDocsTillNow >= k && i+1 >= nprobe {
minEligibleCentroids = i + 1
break
}
}
// Search the clusters specified by 'closestCentroidIDs' for
// vectors whose IDs are present in 'vectorIDsToInclude'
scores, ids, err := vecIndex.SearchClustersFromIVFIndex(
selector, closestCentroidIDs, minEligibleCentroids,
k, qVector, centroidDistances, params)
if err != nil {
return nil, err
}
addIDsToPostingsList(rv, ids, scores)
return rv, nil
},
close: func() {
// skipping the closing because the index is cached and it's being
// deferred to a later point of time.
sb.vecIndexCache.decRef(fieldIDPlus1)
},
size: func() uint64 {
return vecIndexSize
},
obtainKCentroidCardinalitiesFromIVFIndex: func(limit int, descending bool) ([]index.CentroidCardinality, error) {
if vecIndex == nil || !vecIndex.IsIVFIndex() {
return nil, nil
}
cardinalities, centroids, err := vecIndex.ObtainKCentroidCardinalitiesFromIVFIndex(limit, descending)
if err != nil {
return nil, err
}
centroidCardinalities := make([]index.CentroidCardinality, len(cardinalities))
for i, cardinality := range cardinalities {
centroidCardinalities[i] = index.CentroidCardinality{
Centroid: centroids[i],
Cardinality: cardinality,
}
}
return centroidCardinalities, nil
},
}
err error
)
fieldIDPlus1 = sb.fieldsMap[field]
rv := &vectorIndexWrapper{sb: sb}
fieldIDPlus1 := sb.fieldsMap[field]
if fieldIDPlus1 <= 0 {
return wrapVecIndex, nil
return rv, nil
}
rv.fieldIDPlus1 = fieldIDPlus1
vectorSection := sb.fieldsSectionsMap[fieldIDPlus1-1][SectionFaissVectorIndex]
// check if the field has a vector section in the segment.
if vectorSection <= 0 {
return wrapVecIndex, nil
return rv, nil
}
pos := int(vectorSection)
@@ -574,15 +302,19 @@ func (sb *SegmentBase) InterpretVectorIndex(field string, requiresFiltering bool
pos += n
}
vecIndex, vecDocIDMap, docVecIDMap, vectorIDsToExclude, err =
var err error
rv.vecIndex, rv.vecDocIDMap, rv.docVecIDMap, rv.vectorIDsToExclude, err =
sb.vecIndexCache.loadOrCreate(fieldIDPlus1, sb.mem[pos:], requiresFiltering,
except)
if vecIndex != nil {
vecIndexSize = vecIndex.Size()
if err != nil {
return nil, err
}
return wrapVecIndex, err
if rv.vecIndex != nil {
rv.vecIndexSize = rv.vecIndex.Size()
}
return rv, nil
}
func (sb *SegmentBase) UpdateFieldStats(stats segment.FieldStats) {

View File

@@ -0,0 +1,645 @@
// Copyright (c) 2025 Couchbase, Inc.
//
// Licensed 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.
//go:build vectors
// +build vectors
package zap
import (
"encoding/json"
"math"
"slices"
"github.com/RoaringBitmap/roaring/v2/roaring64"
"github.com/bits-and-blooms/bitset"
index "github.com/blevesearch/bleve_index_api"
faiss "github.com/blevesearch/go-faiss"
segment "github.com/blevesearch/scorch_segment_api/v2"
)
// MaxMultiVectorDocSearchRetries limits repeated searches when deduplicating
// multi-vector documents. Each retry excludes previously seen vectors to find
// new unique documents. Acts as a safeguard against pathological data distributions.
var MaxMultiVectorDocSearchRetries = 100
// vectorIndexWrapper conforms to scorch_segment_api's VectorIndex interface
type vectorIndexWrapper struct {
vecIndex *faiss.IndexImpl
vecDocIDMap map[int64]uint32
docVecIDMap map[uint32][]int64
vectorIDsToExclude []int64
fieldIDPlus1 uint16
vecIndexSize uint64
sb *SegmentBase
}
func (v *vectorIndexWrapper) Search(qVector []float32, k int64,
params json.RawMessage) (
segment.VecPostingsList, error) {
// 1. returned postings list (of type PostingsList) has two types of information - docNum and its score.
// 2. both the values can be represented using roaring bitmaps.
// 3. the Iterator (of type PostingsIterator) returned would operate in terms of VecPostings.
// 4. VecPostings would just have the docNum and the score. Every call of Next()
// and Advance just returns the next VecPostings. The caller would do a vp.Number()
// and the Score() to get the corresponding values
rv := &VecPostingsList{
except: nil, // todo: handle the except bitmap within postings iterator.
postings: roaring64.New(),
}
if v.vecIndex == nil || v.vecIndex.D() != len(qVector) {
// vector index not found or dimensionality mismatched
return rv, nil
}
if v.sb.numDocs == 0 {
return rv, nil
}
rs, err := v.searchWithoutIDs(qVector, k,
v.vectorIDsToExclude, params)
if err != nil {
return nil, err
}
v.addIDsToPostingsList(rv, rs)
return rv, nil
}
func (v *vectorIndexWrapper) SearchWithFilter(qVector []float32, k int64,
eligibleDocIDs []uint64, params json.RawMessage) (
segment.VecPostingsList, error) {
// If every element in the index is eligible (full selectivity),
// then this can basically be considered unfiltered kNN.
if len(eligibleDocIDs) == int(v.sb.numDocs) {
return v.Search(qVector, k, params)
}
// 1. returned postings list (of type PostingsList) has two types of information - docNum and its score.
// 2. both the values can be represented using roaring bitmaps.
// 3. the Iterator (of type PostingsIterator) returned would operate in terms of VecPostings.
// 4. VecPostings would just have the docNum and the score. Every call of Next()
// and Advance just returns the next VecPostings. The caller would do a vp.Number()
// and the Score() to get the corresponding values
rv := &VecPostingsList{
except: nil, // todo: handle the except bitmap within postings iterator.
postings: roaring64.New(),
}
if v.vecIndex == nil || v.vecIndex.D() != len(qVector) {
// vector index not found or dimensionality mismatched
return rv, nil
}
// Check and proceed only if non-zero documents eligible per the filter query.
if len(eligibleDocIDs) == 0 {
return rv, nil
}
// vector IDs corresponding to the local doc numbers to be
// considered for the search
vectorIDsToInclude := make([]int64, 0, len(eligibleDocIDs))
for _, id := range eligibleDocIDs {
vecIDs := v.docVecIDMap[uint32(id)]
// In the common case where vecIDs has only one element, which occurs
// when a document has only one vector field, we can
// avoid the unnecessary overhead of slice unpacking (append(vecIDs...)).
// Directly append the single element for efficiency.
if len(vecIDs) == 1 {
vectorIDsToInclude = append(vectorIDsToInclude, vecIDs[0])
} else {
vectorIDsToInclude = append(vectorIDsToInclude, vecIDs...)
}
}
// In case a doc has invalid vector fields but valid non-vector fields,
// filter hit IDs may be ineligible for the kNN since the document does
// not have any/valid vectors.
if len(vectorIDsToInclude) == 0 {
return rv, nil
}
// If the index is not an IVF index, then the search can be
// performed directly, using the Flat index.
if !v.vecIndex.IsIVFIndex() {
// vector IDs corresponding to the local doc numbers to be
// considered for the search
rs, err := v.searchWithIDs(qVector, k,
vectorIDsToInclude, params)
if err != nil {
return nil, err
}
v.addIDsToPostingsList(rv, rs)
return rv, nil
}
// Determining which clusters, identified by centroid ID,
// have at least one eligible vector and hence, ought to be
// probed.
clusterVectorCounts, err := v.vecIndex.ObtainClusterVectorCountsFromIVFIndex(vectorIDsToInclude)
if err != nil {
return nil, err
}
var ids []int64
var include bool
// If there are more elements to be included than excluded, it
// might be quicker to use an exclusion selector as a filter
// instead of an inclusion selector.
if float32(len(eligibleDocIDs))/float32(len(v.docVecIDMap)) > 0.5 {
// Use a bitset to efficiently track eligible document IDs.
// This reduces the lookup cost when checking if a document ID is eligible,
// compared to using a map or slice.
bs := bitset.New(uint(v.sb.numDocs))
for _, docID := range eligibleDocIDs {
bs.Set(uint(docID))
}
ineligibleVectorIDs := make([]int64, 0, len(v.vecDocIDMap)-len(vectorIDsToInclude))
for docID, vecIDs := range v.docVecIDMap {
// Check if the document ID is NOT in the eligible set, marking it as ineligible.
if !bs.Test(uint(docID)) {
// In the common case where vecIDs has only one element, which occurs
// when a document has only one vector field, we can
// avoid the unnecessary overhead of slice unpacking (append(vecIDs...)).
// Directly append the single element for efficiency.
if len(vecIDs) == 1 {
ineligibleVectorIDs = append(ineligibleVectorIDs, vecIDs[0])
} else {
ineligibleVectorIDs = append(ineligibleVectorIDs, vecIDs...)
}
}
}
ids = ineligibleVectorIDs
include = false
} else {
ids = vectorIDsToInclude
include = true
}
// Ordering the retrieved centroid IDs by increasing order
// of distance i.e. decreasing order of proximity to query vector.
centroidIDs := make([]int64, 0, len(clusterVectorCounts))
for centroidID := range clusterVectorCounts {
centroidIDs = append(centroidIDs, centroidID)
}
closestCentroidIDs, centroidDistances, err :=
v.vecIndex.ObtainClustersWithDistancesFromIVFIndex(qVector, centroidIDs)
if err != nil {
return nil, err
}
// Getting the nprobe value set at index time.
nprobe := int(v.vecIndex.GetNProbe())
// Determining the minimum number of centroids to be probed
// to ensure that at least 'k' vectors are collected while
// examining at least 'nprobe' centroids.
// centroidsToProbe range: [nprobe, number of eligible centroids]
var eligibleVecsTillNow int64
centroidsToProbe := len(closestCentroidIDs)
for i, centroidID := range closestCentroidIDs {
eligibleVecsTillNow += clusterVectorCounts[centroidID]
// Stop once we've examined at least 'nprobe' centroids and
// collected at least 'k' vectors.
if eligibleVecsTillNow >= k && i+1 >= nprobe {
centroidsToProbe = i + 1
break
}
}
// Search the clusters specified by 'closestCentroidIDs' for
// vectors whose IDs are present in 'vectorIDsToInclude'
rs, err := v.searchClustersFromIVFIndex(
ids, include, closestCentroidIDs, centroidsToProbe,
k, qVector, centroidDistances, params)
if err != nil {
return nil, err
}
v.addIDsToPostingsList(rv, rs)
return rv, nil
}
func (v *vectorIndexWrapper) Close() {
// skipping the closing because the index is cached and it's being
// deferred to a later point of time.
v.sb.vecIndexCache.decRef(v.fieldIDPlus1)
}
func (v *vectorIndexWrapper) Size() uint64 {
return v.vecIndexSize
}
func (v *vectorIndexWrapper) ObtainKCentroidCardinalitiesFromIVFIndex(limit int, descending bool) (
[]index.CentroidCardinality, error) {
if v.vecIndex == nil || !v.vecIndex.IsIVFIndex() {
return nil, nil
}
cardinalities, centroids, err := v.vecIndex.ObtainKCentroidCardinalitiesFromIVFIndex(limit, descending)
if err != nil {
return nil, err
}
centroidCardinalities := make([]index.CentroidCardinality, len(cardinalities))
for i, cardinality := range cardinalities {
centroidCardinalities[i] = index.CentroidCardinality{
Centroid: centroids[i],
Cardinality: cardinality,
}
}
return centroidCardinalities, nil
}
// Utility function to add the corresponding docID and scores for each unique
// docID retrieved from the vector index search to the newly created vecPostingsList
func (v *vectorIndexWrapper) addIDsToPostingsList(pl *VecPostingsList, rs resultSet) {
rs.iterate(func(docID uint32, score float32) {
// transform the docID and score to vector code format
code := getVectorCode(docID, score)
// add to postings list, this ensures ordered storage
// based on the docID since it occupies the upper 32 bits
pl.postings.Add(code)
})
}
// docSearch performs a search on the vector index to retrieve
// top k documents based on the provided search function.
// It handles deduplication of documents that may have multiple
// vectors associated with them.
// The prepareNextIter function is used to set up the state
// for the next iteration, if more searches are needed to find
// k unique documents. The callback recieves the number of iterations
// done so far and the vector ids retrieved in the last search. While preparing
// the next iteration, if its decided that no further searches are needed,
// the prepareNextIter function can decide whether to continue searching or not
func (v *vectorIndexWrapper) docSearch(k int64, numDocs uint64,
search func() (scores []float32, labels []int64, err error),
prepareNextIter func(numIter int, labels []int64) bool) (resultSet, error) {
// create a result set to hold top K docIDs and their scores
rs := newResultSet(k, numDocs)
// flag to indicate if we have exhausted the vector index
var exhausted bool
// keep track of number of iterations done, we execute the loop more than once only when
// we have multi-vector documents leading to duplicates in docIDs retrieved
numIter := 0
// get the metric type of the index to help with deduplication logic
metricType := v.vecIndex.MetricType()
// we keep searching until we have k unique docIDs or we have exhausted the vector index
// or we have reached the maximum number of deduplication iterations allowed
for numIter < MaxMultiVectorDocSearchRetries && rs.size() < k && !exhausted {
// search the vector index
numIter++
scores, labels, err := search()
if err != nil {
return nil, err
}
// process the retrieved ids and scores, getting the corresponding docIDs
// for each vector id retrieved, and storing the best score for each unique docID
// the moment we see a -1 for a vector id, we stop processing further since
// it indicates there are no more vectors to be retrieved and break out of the loop
// by setting the exhausted flag
for i, vecID := range labels {
if vecID == -1 {
exhausted = true
break
}
docID, exists := v.getDocIDForVectorID(vecID)
if !exists {
continue
}
score := scores[i]
prevScore, exists := rs.get(docID)
if !exists {
// first time seeing this docID, so just store it
rs.put(docID, score)
continue
}
// we have seen this docID before, so we must compare scores
// check the index metric type first to check how we compare distances/scores
// and store the best score for the docID accordingly
// for inner product, higher the score, better the match
// for euclidean distance, lower the score/distance, better the match
// so we invert the comparison accordingly
switch metricType {
case faiss.MetricInnerProduct: // similarity metrics like dot product => higher is better
if score > prevScore {
rs.put(docID, score)
}
case faiss.MetricL2:
fallthrough
default: // distance metrics like euclidean distance => lower is better
if score < prevScore {
rs.put(docID, score)
}
}
}
// if we still have less than k unique docIDs, prepare for the next iteration, provided
// we have not exhausted the index
if rs.size() < k && !exhausted {
// prepare state for next iteration
shouldContinue := prepareNextIter(numIter, labels)
if !shouldContinue {
break
}
}
}
// at this point we either have k unique docIDs or we have exhausted
// the vector index or we have reached the maximum number of deduplication iterations allowed
// or the prepareNextIter function decided to break out of the loop
return rs, nil
}
// searchWithoutIDs performs a search on the vector index to retrieve the top K documents while
// excluding any vector IDs specified in the exclude slice.
func (v *vectorIndexWrapper) searchWithoutIDs(qVector []float32, k int64, exclude []int64, params json.RawMessage) (
resultSet, error) {
return v.docSearch(k, v.sb.numDocs,
func() ([]float32, []int64, error) {
return v.vecIndex.SearchWithoutIDs(qVector, k, exclude, params)
},
func(numIter int, labels []int64) bool {
// if this is the first loop iteration and we have < k unique docIDs,
// we must clone the existing exclude slice before appending to it
// to avoid modifying the original slice passed in by the caller
if numIter == 1 {
exclude = slices.Clone(exclude)
}
// prepare the exclude list for the next iteration by adding
// the vector ids retrieved in this iteration
exclude = append(exclude, labels...)
// with exclude list updated, we can proceed to the next iteration
return true
})
}
// searchWithIDs performs a search on the vector index to retrieve the top K documents while only
// considering the vector IDs specified in the include slice.
func (v *vectorIndexWrapper) searchWithIDs(qVector []float32, k int64, include []int64, params json.RawMessage) (
resultSet, error) {
// if the number of iterations > 1, we will be modifying the include slice
// to exclude vector ids already seen, so we use this set to track the
// include set for the next iteration, this is reused across iterations
// and allocated only once, when numIter == 1
var includeSet map[int64]struct{}
return v.docSearch(k, v.sb.numDocs,
func() ([]float32, []int64, error) {
return v.vecIndex.SearchWithIDs(qVector, k, include, params)
},
func(numIter int, labels []int64) bool {
// if this is the first loop iteration and we have < k unique docIDs,
// we clone the existing include slice before modifying it
if numIter == 1 {
include = slices.Clone(include)
// build the include set for subsequent iterations
includeSet = make(map[int64]struct{}, len(include))
for _, id := range include {
includeSet[id] = struct{}{}
}
}
// prepare the include list for the next iteration
// by removing the vector ids retrieved in this iteration
// from the include set
for _, id := range labels {
delete(includeSet, id)
}
// now build the next include slice from the set
include = include[:0]
for id := range includeSet {
include = append(include, id)
}
// only continue searching if we still have vector ids to include
return len(include) != 0
})
}
// searchClustersFromIVFIndex performs a search on the IVF vector index to retrieve the top K documents
// while either including or excluding the vector IDs specified in the ids slice, depending on the include flag.
// It takes into account the eligible centroid IDs and ensures that at least centroidsToProbe are probed.
// If after a few iterations we haven't found enough documents, it dynamically increases the number of
// clusters searched (up to the number of eligible centroids) to ensure we can find k unique documents.
func (v *vectorIndexWrapper) searchClustersFromIVFIndex(ids []int64, include bool, eligibleCentroidIDs []int64,
centroidsToProbe int, k int64, x, centroidDis []float32, params json.RawMessage) (
resultSet, error) {
// if the number of iterations > 1, we will be modifying the include slice
// to exclude vector ids already seen, so we use this set to track the
// include set for the next iteration, this is reused across iterations
// and allocated only once, when numIter == 1
var includeSet map[int64]struct{}
var totalEligibleCentroids = len(eligibleCentroidIDs)
// Threshold for when to start increasing: after 2 iterations without
// finding enough documents, we start increasing up to the number of centroidsToProbe
// up to the total number of eligible centroids available
const nprobeIncreaseThreshold = 2
return v.docSearch(k, v.sb.numDocs,
func() ([]float32, []int64, error) {
// build the selector based on whatever ids is as of now and the
// include/exclude flag
selector, err := v.getSelector(ids, include)
if err != nil {
return nil, nil, err
}
// once the main search is done we must free the selector
defer selector.Delete()
return v.vecIndex.SearchClustersFromIVFIndex(selector, eligibleCentroidIDs,
centroidsToProbe, k, x, centroidDis, params)
},
func(numIter int, labels []int64) bool {
// if this is the first loop iteration and we have < k unique docIDs,
// we must clone the existing ids slice before modifying it to avoid
// modifying the original slice passed in by the caller
if numIter == 1 {
ids = slices.Clone(ids)
if include {
// build the include set for subsequent iterations
// by adding all the ids initially present in the ids slice
includeSet = make(map[int64]struct{}, len(ids))
for _, id := range ids {
includeSet[id] = struct{}{}
}
}
}
// if we have iterated atleast nprobeIncreaseThreshold times
// and still have not found enough unique docIDs, we increase
// the number of centroids to probe for the next iteration
// to try and find more vectors/documents
if numIter >= nprobeIncreaseThreshold && centroidsToProbe < len(eligibleCentroidIDs) {
// Calculate how much to increase: increase by 50% of the remaining centroids to probe,
// but at least by 1 to ensure progress.
increaseAmount := max((totalEligibleCentroids-centroidsToProbe)/2, 1)
// Update centroidsToProbe, ensuring it does not exceed the total eligible centroids
centroidsToProbe = min(centroidsToProbe+increaseAmount, len(eligibleCentroidIDs))
}
// prepare the exclude/include list for the next iteration
if include {
// removing the vector ids retrieved in this iteration
// from the include set and rebuild the ids slice from the set
for _, id := range labels {
delete(includeSet, id)
}
// now build the next include slice from the set
ids = ids[:0]
for id := range includeSet {
ids = append(ids, id)
}
// only continue searching if we still have vector ids to include
return len(ids) != 0
} else {
// appending the vector ids retrieved in this iteration
// to the exclude list
ids = append(ids, labels...)
// with exclude list updated, we can proceed to the next iteration
return true
}
})
}
// Utility function to get a faiss.Selector based on the include/exclude flag
// and the vector ids provided, if include is true, it returns an inclusion selector,
// else it returns an exclusion selector. The caller must ensure to free the selector
// by calling selector.Delete() when done using it.
func (v *vectorIndexWrapper) getSelector(ids []int64, include bool) (selector faiss.Selector, err error) {
if include {
selector, err = faiss.NewIDSelectorBatch(ids)
} else {
selector, err = faiss.NewIDSelectorNot(ids)
}
if err != nil {
return nil, err
}
return selector, nil
}
// Utility function to get the docID for a given vectorID, used for the
// deduplication logic, to map vectorIDs back to their corresponding docIDs
func (v *vectorIndexWrapper) getDocIDForVectorID(vecID int64) (uint32, bool) {
docID, exists := v.vecDocIDMap[vecID]
return docID, exists
}
// resultSet is a data structure to hold (docID, score) pairs while ensuring
// that each docID is unique. It supports efficient insertion, retrieval,
// and iteration over the stored pairs.
type resultSet interface {
// Add a (docID, score) pair to the result set.
put(docID uint32, score float32)
// Get the score for a given docID. Returns false if docID not present.
get(docID uint32) (float32, bool)
// Iterate over all (docID, score) pairs in the result set.
iterate(func(docID uint32, score float32))
// Get the size of the result set.
size() int64
}
// resultSetSliceThreshold defines the threshold ratio of k to total documents
// in the index, below which a map-based resultSet is used, and above which
// a slice-based resultSet is used.
// It is derived using the following reasoning:
//
// Let N = total number of documents
// Let K = number of top K documents to retrieve
//
// Memory usage if the Result Set uses a map[uint32]float32 of size K underneath:
//
// ~20 bytes per entry (key + value + map overhead)
// Total ≈ 20 * K bytes
//
// Memory usage if the Result Set uses a slice of float32 of size N underneath:
//
// 4 bytes per entry
// Total ≈ 4 * N bytes
//
// We want the threshold below which a map is more memory-efficient than a slice:
//
// 20K < 4N
// K/N < 4/20
//
// Therefore, if the ratio of K to N is less than 0.2 (4/20), we use a map-based resultSet.
const resultSetSliceThreshold float64 = 0.2
// newResultSet creates a new resultSet
func newResultSet(k int64, numDocs uint64) resultSet {
// if numDocs is zero (empty index), just use map-based resultSet as its a no-op
// else decide based the percent of documents being retrieved. If we require
// greater than 20% of total documents, use slice-based resultSet for better memory efficiency
// else use map-based resultSet
if numDocs == 0 || float64(k)/float64(numDocs) < resultSetSliceThreshold {
return newResultSetMap(k)
}
return newResultSetSlice(numDocs)
}
type resultSetMap struct {
data map[uint32]float32
}
func newResultSetMap(k int64) resultSet {
return &resultSetMap{
data: make(map[uint32]float32, k),
}
}
func (rs *resultSetMap) put(docID uint32, score float32) {
rs.data[docID] = score
}
func (rs *resultSetMap) get(docID uint32) (float32, bool) {
score, exists := rs.data[docID]
return score, exists
}
func (rs *resultSetMap) iterate(f func(docID uint32, score float32)) {
for docID, score := range rs.data {
f(docID, score)
}
}
func (rs *resultSetMap) size() int64 {
return int64(len(rs.data))
}
type resultSetSlice struct {
count int64
data []float32
}
func newResultSetSlice(numDocs uint64) resultSet {
data := make([]float32, numDocs)
// scores can be negative, so initialize to a sentinel value which is NaN
sentinel := float32(math.NaN())
for i := range data {
data[i] = sentinel
}
return &resultSetSlice{
count: 0,
data: data,
}
}
func (rs *resultSetSlice) put(docID uint32, score float32) {
// only increment count if this docID was not already present
if math.IsNaN(float64(rs.data[docID])) {
rs.count++
}
rs.data[docID] = score
}
func (rs *resultSetSlice) get(docID uint32) (float32, bool) {
score := rs.data[docID]
if math.IsNaN(float64(score)) {
return 0, false
}
return score, true
}
func (rs *resultSetSlice) iterate(f func(docID uint32, score float32)) {
for docID, score := range rs.data {
if !math.IsNaN(float64(score)) {
f(uint32(docID), score)
}
}
}
func (rs *resultSetSlice) size() int64 {
return rs.count
}

4
vendor/modules.txt vendored
View File

@@ -117,7 +117,7 @@ github.com/bitly/go-simplejson
# github.com/bits-and-blooms/bitset v1.22.0
## explicit; go 1.16
github.com/bits-and-blooms/bitset
# github.com/blevesearch/bleve/v2 v2.5.5
# github.com/blevesearch/bleve/v2 v2.5.7
## explicit; go 1.23
github.com/blevesearch/bleve/v2
github.com/blevesearch/bleve/v2/analysis
@@ -217,7 +217,7 @@ github.com/blevesearch/zapx/v14
# github.com/blevesearch/zapx/v15 v15.4.2
## explicit; go 1.21
github.com/blevesearch/zapx/v15
# github.com/blevesearch/zapx/v16 v16.2.7
# github.com/blevesearch/zapx/v16 v16.2.8
## explicit; go 1.23
github.com/blevesearch/zapx/v16
# github.com/bluele/gcache v0.0.2