wehub-resource-sync
498b235461
Build and test / Build and test AMD64 Ubuntu 22.04 (push) Failing after 0s
Publish Builder / amazonlinux2023 (push) Failing after 1s
Build and test / UT for Go (push) Has been skipped
Publish KRTE Images / KRTE (push) Failing after 1s
Build and test / Integration Test (push) Has been skipped
Build and test / Upload Code Coverage (push) Has been skipped
Publish Builder / rockylinux9 (push) Failing after 1s
Publish Builder / ubuntu22.04 (push) Failing after 0s
Publish Builder / ubuntu24.04 (push) Failing after 0s
Publish Gpu Builder / publish-gpu-builder (push) Failing after 1s
Publish Test Images / PyTest (push) Failing after 0s
Build and test / UT for Cpp (push) Has been cancelled
1514 lines
50 KiB
Go
1514 lines
50 KiB
Go
// Licensed to the LF AI & Data foundation under one
|
|
// or more contributor license agreements. See the NOTICE file
|
|
// distributed with this work for additional information
|
|
// regarding copyright ownership. The ASF licenses this file
|
|
// to you under the Apache License, Version 2.0 (the
|
|
// "License"); you may not use this file except in compliance
|
|
// with the License. You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package delegator
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"runtime"
|
|
"time"
|
|
|
|
"github.com/cockroachdb/errors"
|
|
"github.com/samber/lo"
|
|
"golang.org/x/sync/errgroup"
|
|
"google.golang.org/protobuf/proto"
|
|
|
|
"github.com/milvus-io/milvus-proto/go-api/v3/commonpb"
|
|
"github.com/milvus-io/milvus-proto/go-api/v3/milvuspb"
|
|
"github.com/milvus-io/milvus-proto/go-api/v3/msgpb"
|
|
"github.com/milvus-io/milvus-proto/go-api/v3/schemapb"
|
|
"github.com/milvus-io/milvus/internal/querynodev2/cluster"
|
|
"github.com/milvus-io/milvus/internal/querynodev2/delegator/deletebuffer"
|
|
"github.com/milvus-io/milvus/internal/querynodev2/pkoracle"
|
|
"github.com/milvus-io/milvus/internal/querynodev2/segments"
|
|
"github.com/milvus-io/milvus/internal/storage"
|
|
"github.com/milvus-io/milvus/internal/storagev2/packed"
|
|
"github.com/milvus-io/milvus/internal/util/function"
|
|
"github.com/milvus-io/milvus/internal/util/grpcclient"
|
|
"github.com/milvus-io/milvus/pkg/v3/common"
|
|
"github.com/milvus-io/milvus/pkg/v3/metrics"
|
|
"github.com/milvus-io/milvus/pkg/v3/mlog"
|
|
"github.com/milvus-io/milvus/pkg/v3/proto/datapb"
|
|
"github.com/milvus-io/milvus/pkg/v3/proto/internalpb"
|
|
"github.com/milvus-io/milvus/pkg/v3/proto/querypb"
|
|
"github.com/milvus-io/milvus/pkg/v3/proto/segcorepb"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/commonpbutil"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/conc"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/funcutil"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/merr"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/retry"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/timerecord"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/tsoutil"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
|
|
)
|
|
|
|
// delegator data related part
|
|
|
|
const defaultAnalyzerName = "default"
|
|
|
|
func normalizeAnalyzerNames(analyzerNames []string, textNum int) ([]string, error) {
|
|
if textNum == 0 {
|
|
return []string{}, nil
|
|
}
|
|
|
|
switch len(analyzerNames) {
|
|
case 0:
|
|
names := make([]string, textNum)
|
|
for i := range names {
|
|
names[i] = defaultAnalyzerName
|
|
}
|
|
return names, nil
|
|
case 1:
|
|
name := analyzerNames[0]
|
|
if name == "" {
|
|
name = defaultAnalyzerName
|
|
}
|
|
names := make([]string, textNum)
|
|
for i := range names {
|
|
names[i] = name
|
|
}
|
|
return names, nil
|
|
case textNum:
|
|
names := append([]string(nil), analyzerNames...)
|
|
for i, name := range names {
|
|
if name == "" {
|
|
names[i] = defaultAnalyzerName
|
|
}
|
|
}
|
|
return names, nil
|
|
default:
|
|
return nil, merr.WrapErrParameterInvalidMsg("analyzer names size must be 0, 1, or equal to text size, got analyzer names size [%d], text size [%d]", len(analyzerNames), textNum)
|
|
}
|
|
}
|
|
|
|
func normalizeHighlightAnalyzerNames(analyzerNames []string, textNum int) ([]string, error) {
|
|
if len(analyzerNames) != textNum {
|
|
return nil, merr.WrapErrServiceInternalMsg("highlight analyzer names size must equal text size, got analyzer names size [%d], text size [%d]", len(analyzerNames), textNum)
|
|
}
|
|
|
|
names := append([]string(nil), analyzerNames...)
|
|
for i, name := range names {
|
|
if name == "" {
|
|
names[i] = defaultAnalyzerName
|
|
}
|
|
}
|
|
return names, nil
|
|
}
|
|
|
|
// segmentEffectiveTs returns the timestamp for delete-buffer pin/ListAfter.
|
|
// For import segments with commit_timestamp, only deletes from T_commit onwards
|
|
// are applied via the buffer (pre-commit deletes are in the delta log or L0 path).
|
|
func segmentEffectiveTs(info *querypb.SegmentLoadInfo) uint64 {
|
|
if ts := info.GetCommitTimestamp(); ts != 0 {
|
|
return ts
|
|
}
|
|
return info.GetStartPosition().GetTimestamp()
|
|
}
|
|
|
|
// InsertData
|
|
type InsertData struct {
|
|
RowIDs []int64
|
|
PrimaryKeys []storage.PrimaryKey
|
|
Timestamps []uint64
|
|
InsertRecord *segcorepb.InsertRecord
|
|
BM25Stats map[int64]*storage.BM25Stats
|
|
|
|
StartPosition *msgpb.MsgPosition
|
|
PartitionID int64
|
|
}
|
|
|
|
type DeleteData struct {
|
|
PartitionID int64
|
|
PrimaryKeys []storage.PrimaryKey
|
|
Timestamps []uint64
|
|
RowCount int64
|
|
}
|
|
|
|
type DeleteBatch struct {
|
|
Ts uint64
|
|
Data []*DeleteData
|
|
}
|
|
|
|
// Append appends another delete data into this one.
|
|
func (d *DeleteData) Append(ad DeleteData) {
|
|
d.PrimaryKeys = append(d.PrimaryKeys, ad.PrimaryKeys...)
|
|
d.Timestamps = append(d.Timestamps, ad.Timestamps...)
|
|
d.RowCount += ad.RowCount
|
|
}
|
|
|
|
// ProcessInsert handles insert data in delegator.
|
|
func (sd *shardDelegator) ProcessInsert(insertRecords map[int64]*InsertData) {
|
|
method := "ProcessInsert"
|
|
tr := timerecord.NewTimeRecorder(method)
|
|
log := sd.getLogger(context.Background())
|
|
for segmentID, insertData := range insertRecords {
|
|
growing := sd.segmentManager.GetGrowing(segmentID)
|
|
newGrowingSegment := false
|
|
if growing == nil {
|
|
var err error
|
|
// TODO: It's a wired implementation that growing segment have load info.
|
|
// we should separate the growing segment and sealed segment by type system.
|
|
growing, err = segments.NewSegment(
|
|
context.Background(),
|
|
sd.collection,
|
|
sd.segmentManager,
|
|
segments.SegmentTypeGrowing,
|
|
0,
|
|
&querypb.SegmentLoadInfo{
|
|
SegmentID: segmentID,
|
|
PartitionID: insertData.PartitionID,
|
|
CollectionID: sd.collectionID,
|
|
InsertChannel: sd.vchannelName,
|
|
StartPosition: insertData.StartPosition,
|
|
DeltaPosition: insertData.StartPosition,
|
|
Level: datapb.SegmentLevel_L1,
|
|
},
|
|
)
|
|
if err != nil {
|
|
log.Error(context.TODO(), "failed to create new segment",
|
|
mlog.FieldSegmentID(segmentID),
|
|
mlog.Err(err))
|
|
panic(err)
|
|
}
|
|
newGrowingSegment = true
|
|
}
|
|
|
|
err := growing.Insert(context.Background(), insertData.RowIDs, insertData.Timestamps, insertData.InsertRecord)
|
|
if err != nil {
|
|
log.Error(context.TODO(), "failed to insert data into growing segment",
|
|
mlog.FieldSegmentID(segmentID),
|
|
mlog.Err(err),
|
|
)
|
|
if errors.IsAny(err, merr.ErrSegmentNotLoaded, merr.ErrSegmentNotFound) {
|
|
log.Warn(context.TODO(), "try to insert data into released segment, skip it", mlog.Err(err))
|
|
continue
|
|
}
|
|
// panic here, insert failure
|
|
panic(err)
|
|
}
|
|
growing.UpdatePkCandidate(insertData.PrimaryKeys)
|
|
|
|
if newGrowingSegment {
|
|
sd.growingSegmentLock.Lock()
|
|
// Forbid create growing segment in excluded segment
|
|
// (Now excluded ts may not worked for growing segment with multiple partition.
|
|
// Because use checkpoint ts as excluded ts when add excluded, but it may less than last message ts.
|
|
// And cause some invalid message not filtered out and create growing again.
|
|
// So we forbid all segment in excluded segment create here.)
|
|
// TODO:
|
|
// Use right ts when add excluded segment. And Verify with insert ts here.
|
|
if ok := sd.VerifyExcludedSegments(segmentID, 0); !ok {
|
|
log.Warn(context.TODO(), "try to insert data into released segment, skip it", mlog.FieldSegmentID(segmentID))
|
|
sd.growingSegmentLock.Unlock()
|
|
growing.Release(context.Background())
|
|
continue
|
|
}
|
|
|
|
if !sd.distribution.GrowingSegmentExists(segmentID) {
|
|
// register created growing segment after insert, avoid to add empty growing to delegator
|
|
if idfOracle := sd.getIDFOracle(); idfOracle != nil {
|
|
idfOracle.RegisterGrowing(segmentID, insertData.BM25Stats)
|
|
}
|
|
sd.segmentManager.Put(context.Background(), segments.SegmentTypeGrowing, growing)
|
|
sd.addGrowing(SegmentEntry{
|
|
NodeID: paramtable.GetNodeID(),
|
|
SegmentID: segmentID,
|
|
PartitionID: insertData.PartitionID,
|
|
Version: 0,
|
|
TargetVersion: initialTargetVersion,
|
|
Candidate: growing, // growing segment itself is the Candidate
|
|
})
|
|
}
|
|
|
|
sd.growingSegmentLock.Unlock()
|
|
} else if idfOracle := sd.getIDFOracle(); idfOracle != nil {
|
|
idfOracle.UpdateGrowing(growing.ID(), insertData.BM25Stats)
|
|
}
|
|
log.Info(context.TODO(), "insert into growing segment",
|
|
mlog.FieldCollectionID(growing.Collection()),
|
|
mlog.FieldSegmentID(segmentID),
|
|
mlog.Int("rowCount", len(insertData.RowIDs)),
|
|
mlog.Uint64("maxTimestamp", insertData.Timestamps[len(insertData.Timestamps)-1]),
|
|
)
|
|
}
|
|
metrics.QueryNodeProcessCost.WithLabelValues(paramtable.GetStringNodeID(), metrics.InsertLabel).
|
|
Observe(float64(tr.ElapseSpan().Milliseconds()))
|
|
}
|
|
|
|
// ProcessDelete handles delete data in delegator.
|
|
// delegator puts deleteData into buffer first,
|
|
// then dispatch data to segments according to the result of bloom filter check.
|
|
func (sd *shardDelegator) ProcessDelete(deleteData []*DeleteData, ts uint64) {
|
|
sd.ProcessDeleteBatches([]DeleteBatch{{Ts: ts, Data: deleteData}})
|
|
}
|
|
|
|
func (sd *shardDelegator) ProcessDeleteBatches(batches []DeleteBatch) {
|
|
// Early return if delegator is stopped - ProcessDelete becomes a no-op
|
|
// This prevents unnecessary processing and side effects during shutdown
|
|
if sd.Stopped() {
|
|
return
|
|
}
|
|
if len(batches) == 0 {
|
|
return
|
|
}
|
|
|
|
method := "ProcessDelete"
|
|
tr := timerecord.NewTimeRecorder(method)
|
|
// block load segment handle delete buffer
|
|
sd.deleteMut.Lock()
|
|
defer sd.deleteMut.Unlock()
|
|
|
|
log := sd.getLogger(context.Background())
|
|
|
|
log.Debug(context.TODO(), "start to process delete batches", mlog.Int("batchNum", len(batches)))
|
|
allDeleteData := make([]*DeleteData, 0, len(batches))
|
|
for _, batch := range batches {
|
|
if len(batch.Data) == 0 {
|
|
continue
|
|
}
|
|
|
|
cacheItems := make([]deletebuffer.BufferItem, 0, len(batch.Data))
|
|
for _, entry := range batch.Data {
|
|
cacheItems = append(cacheItems, deletebuffer.BufferItem{
|
|
PartitionID: entry.PartitionID,
|
|
DeleteData: storage.DeleteData{
|
|
Pks: entry.PrimaryKeys,
|
|
Tss: entry.Timestamps,
|
|
RowCount: entry.RowCount,
|
|
},
|
|
})
|
|
}
|
|
|
|
sd.deleteBuffer.Put(&deletebuffer.Item{
|
|
Ts: batch.Ts,
|
|
Data: cacheItems,
|
|
})
|
|
allDeleteData = append(allDeleteData, batch.Data...)
|
|
}
|
|
|
|
if len(allDeleteData) > 0 {
|
|
sd.forwardStreamingDeletion(context.Background(), allDeleteData)
|
|
}
|
|
|
|
metrics.QueryNodeProcessCost.WithLabelValues(paramtable.GetStringNodeID(), metrics.DeleteLabel).
|
|
Observe(float64(tr.ElapseSpan().Milliseconds()))
|
|
}
|
|
|
|
type BatchApplyRet = struct {
|
|
DeleteDataIdx int
|
|
StartIdx int
|
|
Segment2Hits map[int64][]bool
|
|
}
|
|
|
|
// applyBFInParallel applies bloom filter check in parallel on the provided pinned segments.
|
|
// Using pinned segments ensures consistency between BF check and delete application,
|
|
// preventing race conditions where new segments could be added between PinOnlineSegments
|
|
// and this call.
|
|
func (sd *shardDelegator) applyBFInParallel(deleteDatas []*DeleteData, pool *conc.Pool[any], sealed []SnapshotItem, growing []SegmentEntry) *typeutil.ConcurrentMap[int, *BatchApplyRet] {
|
|
retIdx := 0
|
|
retMap := typeutil.NewConcurrentMap[int, *BatchApplyRet]()
|
|
batchSize := paramtable.Get().CommonCfg.BloomFilterApplyBatchSize.GetAsInt()
|
|
|
|
var futures []*conc.Future[any]
|
|
for didx, data := range deleteDatas {
|
|
pks := data.PrimaryKeys
|
|
for idx := 0; idx < len(pks); idx += batchSize {
|
|
startIdx := idx
|
|
endIdx := startIdx + batchSize
|
|
if endIdx > len(pks) {
|
|
endIdx = len(pks)
|
|
}
|
|
|
|
retIdx += 1
|
|
tmpRetIndex := retIdx
|
|
deleteDataId := didx
|
|
partitionID := data.PartitionID
|
|
future := pool.Submit(func() (any, error) {
|
|
ret := BatchGetFromSegments(pks[startIdx:endIdx], partitionID, sealed, growing)
|
|
retMap.Insert(tmpRetIndex, &BatchApplyRet{
|
|
DeleteDataIdx: deleteDataId,
|
|
StartIdx: startIdx,
|
|
Segment2Hits: ret,
|
|
})
|
|
return nil, nil
|
|
})
|
|
futures = append(futures, future)
|
|
}
|
|
}
|
|
conc.AwaitAll(futures...)
|
|
|
|
return retMap
|
|
}
|
|
|
|
// applyDelete handles delete record and apply them to corresponding workers.
|
|
func (sd *shardDelegator) applyDelete(ctx context.Context,
|
|
nodeID int64,
|
|
worker cluster.Worker,
|
|
delRecords func(segmentID int64) (DeleteData, bool),
|
|
entries []SegmentEntry,
|
|
scope querypb.DataScope,
|
|
) []int64 {
|
|
offlineSegments := typeutil.NewConcurrentSet[int64]()
|
|
log := sd.getLogger(ctx)
|
|
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
defer cancel()
|
|
|
|
pool := conc.NewPool[struct{}](runtime.GOMAXPROCS(0) * 4)
|
|
defer pool.Release()
|
|
|
|
var futures []*conc.Future[struct{}]
|
|
for _, segmentEntry := range entries {
|
|
segmentEntry := segmentEntry
|
|
delRecord, ok := delRecords(segmentEntry.SegmentID)
|
|
log := log.With(
|
|
mlog.FieldSegmentID(segmentEntry.SegmentID),
|
|
mlog.Int64("workerID", nodeID),
|
|
mlog.Int("forwardRowCount", len(delRecord.PrimaryKeys)),
|
|
)
|
|
if ok {
|
|
future := pool.Submit(func() (struct{}, error) {
|
|
log.Debug(ctx, "delegator plan to applyDelete via worker")
|
|
err := retry.Handle(ctx, func() (bool, error) {
|
|
if sd.Stopped() {
|
|
return false, merr.WrapErrChannelNotAvailable(sd.vchannelName, "channel is unsubscribing")
|
|
}
|
|
|
|
err := worker.Delete(ctx, &querypb.DeleteRequest{
|
|
Base: commonpbutil.NewMsgBase(commonpbutil.WithTargetID(nodeID)),
|
|
CollectionId: sd.collectionID,
|
|
PartitionId: segmentEntry.PartitionID,
|
|
VchannelName: sd.vchannelName,
|
|
SegmentId: segmentEntry.SegmentID,
|
|
PrimaryKeys: storage.ParsePrimaryKeys2IDs(delRecord.PrimaryKeys),
|
|
Timestamps: delRecord.Timestamps,
|
|
Scope: scope,
|
|
})
|
|
if errors.Is(err, merr.ErrNodeNotFound) {
|
|
log.Warn(ctx, "try to delete data on non-exist node")
|
|
// cancel other request
|
|
cancel()
|
|
return false, err
|
|
} else if grpcclient.IsServerIDMismatchErr(err) {
|
|
log.Warn(ctx, "try to delete data on mismatched node, node has been replaced", mlog.Err(err))
|
|
cancel()
|
|
return false, err
|
|
} else if errors.IsAny(err, merr.ErrSegmentNotFound, merr.ErrSegmentNotLoaded) {
|
|
log.Warn(ctx, "try to delete data of released segment")
|
|
return false, nil
|
|
} else if err != nil {
|
|
log.Warn(ctx, "worker failed to delete on segment", mlog.Err(err))
|
|
return true, err
|
|
}
|
|
return false, nil
|
|
}, retry.Attempts(10))
|
|
if err != nil {
|
|
log.Warn(ctx, "apply delete for segment failed, marking it offline")
|
|
offlineSegments.Insert(segmentEntry.SegmentID)
|
|
}
|
|
return struct{}{}, err
|
|
})
|
|
futures = append(futures, future)
|
|
}
|
|
}
|
|
conc.AwaitAll(futures...)
|
|
return offlineSegments.Collect()
|
|
}
|
|
|
|
// markSegmentOffline makes segment go offline and waits for QueryCoord to fix.
|
|
func (sd *shardDelegator) markSegmentOffline(segmentIDs ...int64) {
|
|
sd.distribution.MarkOfflineSegments(segmentIDs...)
|
|
}
|
|
|
|
// addGrowing add growing segment record for delegator.
|
|
func (sd *shardDelegator) addGrowing(entries ...SegmentEntry) {
|
|
log := sd.getLogger(context.Background())
|
|
log.Info(context.TODO(), "add growing segments to delegator", mlog.Int64s("segmentIDs", lo.Map(entries, func(entry SegmentEntry, _ int) int64 {
|
|
return entry.SegmentID
|
|
})))
|
|
sd.distribution.AddGrowing(entries...)
|
|
}
|
|
|
|
// LoadGrowing load growing segments locally.
|
|
func (sd *shardDelegator) LoadGrowing(ctx context.Context, infos []*querypb.SegmentLoadInfo, version int64) error {
|
|
log := sd.getLogger(ctx)
|
|
|
|
segmentIDs := lo.Map(infos, func(info *querypb.SegmentLoadInfo, _ int) int64 { return info.GetSegmentID() })
|
|
log.Info(ctx, "loading growing segments...", mlog.Int64s("segmentIDs", segmentIDs))
|
|
loaded, err := sd.loader.Load(ctx, sd.collectionID, segments.SegmentTypeGrowing, version, infos...)
|
|
if err != nil {
|
|
log.Warn(ctx, "failed to load growing segment", mlog.Err(err))
|
|
return err
|
|
}
|
|
|
|
for _, segment := range loaded {
|
|
err = sd.addL0ForGrowing(ctx, segment)
|
|
if err != nil {
|
|
log.Warn(ctx, "failed to forward L0 deletions to growing segment",
|
|
mlog.Err(err),
|
|
)
|
|
|
|
// clear loaded growing segments
|
|
for _, segment := range loaded {
|
|
segment.Release(ctx)
|
|
}
|
|
return err
|
|
}
|
|
}
|
|
|
|
segmentIDs = lo.Map(loaded, func(segment segments.Segment, _ int) int64 { return segment.ID() })
|
|
log.Info(ctx, "load growing segments done", mlog.Int64s("segmentIDs", segmentIDs))
|
|
|
|
if idfOracle := sd.getIDFOracle(); idfOracle != nil {
|
|
for _, segment := range loaded {
|
|
idfOracle.RegisterGrowing(segment.ID(), segment.GetBM25Stats())
|
|
}
|
|
}
|
|
sd.addGrowing(lo.Map(loaded, func(segment segments.Segment, _ int) SegmentEntry {
|
|
return SegmentEntry{
|
|
NodeID: paramtable.GetNodeID(),
|
|
SegmentID: segment.ID(),
|
|
PartitionID: segment.Partition(),
|
|
Version: version,
|
|
TargetVersion: sd.distribution.getTargetVersion(),
|
|
Candidate: segment, // growing segment itself is the Candidate
|
|
}
|
|
})...)
|
|
return nil
|
|
}
|
|
|
|
// load bm25 stats for sealed segments.
|
|
// idf oracle owns the full lifecycle: download, disk write, register, cleanup.
|
|
func (sd *shardDelegator) loadBM25Stats(ctx context.Context, infos []*querypb.SegmentLoadInfo, req *querypb.LoadSegmentsRequest) error {
|
|
idfOracle := sd.getIDFOracle()
|
|
if idfOracle == nil {
|
|
return nil
|
|
}
|
|
|
|
pool := segments.GetLoadPool()
|
|
|
|
cm := sd.loader.GetChunkManager()
|
|
futures := make([]*conc.Future[any], 0, len(infos))
|
|
for _, info := range infos {
|
|
info := info
|
|
futures = append(futures, pool.Submit(func() (any, error) {
|
|
if err := idfOracle.LoadSealed(ctx, info.GetSegmentID(), info, cm); err != nil {
|
|
mlog.Warn(ctx, "failed to load bm25 stats for segment",
|
|
mlog.FieldCollectionID(req.GetCollectionID()),
|
|
mlog.FieldSegmentID(info.GetSegmentID()),
|
|
mlog.Err(err))
|
|
return nil, err
|
|
}
|
|
return nil, nil
|
|
}))
|
|
}
|
|
|
|
err := conc.BlockOnAll(futures...)
|
|
if err != nil {
|
|
mlog.Warn(ctx, "failed to load bm25 stats", mlog.Err(err))
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (sd *shardDelegator) handleReopenPostLoad(ctx context.Context, req *querypb.LoadSegmentsRequest) error {
|
|
log := sd.getLogger(ctx).With(
|
|
mlog.Int64s("segments", lo.Map(req.GetInfos(), func(info *querypb.SegmentLoadInfo, _ int) int64 { return info.GetSegmentID() })),
|
|
mlog.String("loadScope", req.GetLoadScope().String()),
|
|
)
|
|
|
|
infosWithBM25Stats := make([]*querypb.SegmentLoadInfo, 0, len(req.GetInfos()))
|
|
for _, info := range req.GetInfos() {
|
|
bm25Paths, err := packed.NewStatsResolverFromLoadInfo(info).BM25StatsPaths()
|
|
if err != nil {
|
|
log.Warn(ctx, "resolve reopened bm25 stats failed", mlog.FieldSegmentID(info.GetSegmentID()), mlog.Err(err))
|
|
return err
|
|
}
|
|
if len(bm25Paths) > 0 {
|
|
infosWithBM25Stats = append(infosWithBM25Stats, info)
|
|
}
|
|
}
|
|
|
|
if len(infosWithBM25Stats) == 0 {
|
|
return nil
|
|
}
|
|
return sd.loadBM25StatsForReopen(ctx, infosWithBM25Stats, req)
|
|
}
|
|
|
|
func (sd *shardDelegator) loadBM25StatsForReopen(ctx context.Context, infos []*querypb.SegmentLoadInfo, req *querypb.LoadSegmentsRequest) error {
|
|
idfOracle := sd.getIDFOracle()
|
|
if idfOracle == nil {
|
|
return merr.WrapErrServiceInternal("reopen contains BM25 stats before delegator BM25 oracle is initialized")
|
|
}
|
|
|
|
pool := segments.GetLoadPool()
|
|
cm := sd.loader.GetChunkManager()
|
|
futures := make([]*conc.Future[any], 0, len(infos))
|
|
for _, info := range infos {
|
|
info := info
|
|
futures = append(futures, pool.Submit(func() (any, error) {
|
|
activateIfReadable := sd.distribution.IsReadableSealedSegment(info.GetSegmentID())
|
|
if err := idfOracle.LoadSealedForReopen(ctx, info.GetSegmentID(), info, cm, activateIfReadable); err != nil {
|
|
mlog.Warn(ctx, "failed to load reopened bm25 stats for segment",
|
|
mlog.FieldCollectionID(req.GetCollectionID()),
|
|
mlog.FieldSegmentID(info.GetSegmentID()),
|
|
mlog.Bool("activateIfReadable", activateIfReadable),
|
|
mlog.Err(err))
|
|
return nil, err
|
|
}
|
|
return nil, nil
|
|
}))
|
|
}
|
|
|
|
if err := conc.BlockOnAll(futures...); err != nil {
|
|
mlog.Warn(ctx, "failed to load reopened bm25 stats", mlog.Err(err))
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// syncCollectionIndexMeta refreshes the delegator node's CCollection IndexMeta after a
|
|
// forwarded worker load. Worker LoadSegments already updates IndexMeta on the target
|
|
// worker, but the delegator (which executes growing search locally) must stay in sync.
|
|
func (sd *shardDelegator) syncCollectionIndexMeta(ctx context.Context, req *querypb.LoadSegmentsRequest) error {
|
|
if len(req.GetIndexInfoList()) == 0 {
|
|
return nil
|
|
}
|
|
|
|
schema := req.GetSchema()
|
|
if schema == nil {
|
|
schema = sd.collection.Schema()
|
|
}
|
|
|
|
loadMeta := req.GetLoadMeta()
|
|
if loadMeta == nil {
|
|
loadMeta = &querypb.LoadMetaInfo{
|
|
CollectionID: req.GetCollectionID(),
|
|
}
|
|
}
|
|
|
|
meta := segments.ComposeIndexMeta(ctx, req.GetIndexInfoList(), schema)
|
|
if err := sd.collectionManager.PutOrRef(req.GetCollectionID(), schema, meta, loadMeta); err != nil {
|
|
return err
|
|
}
|
|
sd.collectionManager.Unref(req.GetCollectionID(), 1)
|
|
return nil
|
|
}
|
|
|
|
// LoadSegments load segments local or remotely depends on the target node.
|
|
func (sd *shardDelegator) LoadSegments(ctx context.Context, req *querypb.LoadSegmentsRequest) error {
|
|
if len(req.GetInfos()) == 0 {
|
|
return nil
|
|
}
|
|
|
|
log := sd.getLogger(ctx)
|
|
|
|
targetNodeID := req.GetDstNodeID()
|
|
// add common log fields
|
|
log = log.With(
|
|
mlog.Int64("workID", targetNodeID),
|
|
mlog.Int64s("segments", lo.Map(req.GetInfos(), func(info *querypb.SegmentLoadInfo, _ int) int64 { return info.GetSegmentID() })),
|
|
)
|
|
|
|
if req.GetInfos()[0].GetLevel() == datapb.SegmentLevel_L0 {
|
|
return merr.WrapErrServiceInternal("load L0 segment is not supported, l0 segment should only be loaded by watchChannel")
|
|
}
|
|
|
|
// pin all segments to prevent delete buffer has been cleaned up during worker load segments
|
|
// Note: if delete records is pinned, it will skip cleanup during SyncTargetVersion
|
|
// which means after segment is loaded, then delete buffer will be cleaned up by next SyncTargetVersion call
|
|
for _, info := range req.GetInfos() {
|
|
sd.deleteBuffer.Pin(segmentEffectiveTs(info), info.GetSegmentID())
|
|
}
|
|
defer func() {
|
|
for _, info := range req.GetInfos() {
|
|
sd.deleteBuffer.Unpin(segmentEffectiveTs(info), info.GetSegmentID())
|
|
}
|
|
}()
|
|
|
|
worker, err := sd.workerManager.GetWorker(ctx, targetNodeID)
|
|
if err != nil {
|
|
log.Warn(ctx, "delegator failed to find worker", mlog.Err(err))
|
|
return err
|
|
}
|
|
|
|
req.Base.TargetID = targetNodeID
|
|
log.Debug(ctx, "worker loads segments...")
|
|
|
|
sLoad := func(ctx context.Context, req *querypb.LoadSegmentsRequest) error {
|
|
segmentID := req.GetInfos()[0].GetSegmentID()
|
|
nodeID := req.GetDstNodeID()
|
|
_, err, _ := sd.sf.Do(fmt.Sprintf("%d-%d", nodeID, segmentID), func() (struct{}, error) {
|
|
err := worker.LoadSegments(ctx, req)
|
|
return struct{}{}, err
|
|
})
|
|
return err
|
|
}
|
|
|
|
// separate infos into different load task
|
|
if len(req.GetInfos()) > 1 {
|
|
var reqs []*querypb.LoadSegmentsRequest
|
|
for _, info := range req.GetInfos() {
|
|
newReq := typeutil.Clone(req)
|
|
newReq.Infos = []*querypb.SegmentLoadInfo{info}
|
|
reqs = append(reqs, newReq)
|
|
}
|
|
|
|
group, ctx := errgroup.WithContext(ctx)
|
|
for _, req := range reqs {
|
|
req := req
|
|
group.Go(func() error {
|
|
return sLoad(ctx, req)
|
|
})
|
|
}
|
|
err = group.Wait()
|
|
} else {
|
|
err = sLoad(ctx, req)
|
|
}
|
|
|
|
if err != nil {
|
|
log.Warn(ctx, "worker failed to load segments", mlog.Err(err))
|
|
return err
|
|
}
|
|
log.Debug(ctx, "work loads segments done")
|
|
|
|
if err := sd.syncCollectionIndexMeta(ctx, req); err != nil {
|
|
log.Warn(ctx, "failed to sync collection index meta on delegator", mlog.Err(err))
|
|
return err
|
|
}
|
|
|
|
// load index segment need no stream delete and distribution change
|
|
if req.GetLoadScope() == querypb.LoadScope_Index {
|
|
return nil
|
|
}
|
|
if req.GetLoadScope() == querypb.LoadScope_Reopen {
|
|
return sd.handleReopenPostLoad(ctx, req)
|
|
}
|
|
|
|
return sd.withPostLoadLimit(ctx, func() error {
|
|
infos := lo.Filter(req.GetInfos(), func(info *querypb.SegmentLoadInfo, _ int) bool {
|
|
return !sd.distribution.SealedSegmentExistsOnNode(info.GetSegmentID(), targetNodeID)
|
|
})
|
|
|
|
candidates, err := sd.loader.LoadBloomFilterSet(ctx, req.GetCollectionID(), infos...)
|
|
if err != nil {
|
|
log.Warn(ctx, "failed to load bloom filter set for segment", mlog.Err(err))
|
|
return err
|
|
}
|
|
|
|
// Load BM25 stats BEFORE loadStreamDelete so stats are ready before segment becomes visible
|
|
err = sd.loadBM25Stats(ctx, infos, req)
|
|
if err != nil {
|
|
log.Warn(ctx, "failed to load BM25 stats", mlog.Err(err))
|
|
return err
|
|
}
|
|
|
|
// Build a map from segmentID to BloomFilterSet
|
|
bfMap := make(map[int64]pkoracle.Candidate)
|
|
for _, candidate := range candidates {
|
|
log.Info(ctx, "loaded bloom filter set for sealed segment",
|
|
mlog.FieldSegmentID(candidate.ID()),
|
|
)
|
|
bfMap[candidate.ID()] = candidate
|
|
}
|
|
|
|
// Build entries with Candidate before loadStreamDelete, which will atomically add them to distribution
|
|
entries := make([]SegmentEntry, 0, len(infos))
|
|
for _, info := range infos {
|
|
entries = append(entries, SegmentEntry{
|
|
SegmentID: info.GetSegmentID(),
|
|
PartitionID: info.GetPartitionID(),
|
|
NodeID: req.GetDstNodeID(),
|
|
Version: req.GetVersion(),
|
|
Level: info.GetLevel(),
|
|
Candidate: bfMap[info.GetSegmentID()],
|
|
})
|
|
}
|
|
|
|
log.Debug(ctx, "load delete...")
|
|
// loadStreamDelete now handles distribution add atomically in Phase 3
|
|
err = sd.loadStreamDelete(ctx, candidates, infos, req, targetNodeID, worker,
|
|
entries, req.GetLoadMeta().GetSchemaBarrierTs())
|
|
if err != nil {
|
|
log.Warn(ctx, "load stream delete failed", mlog.Err(err))
|
|
// BM25 stats already loaded into idf oracle will be cleaned up
|
|
// automatically by SyncDistribution when the segment is not in target.
|
|
return err
|
|
}
|
|
log.Debug(ctx, "load stream delete done")
|
|
|
|
return nil
|
|
})
|
|
}
|
|
|
|
func (sd *shardDelegator) withPostLoadLimit(ctx context.Context, fn func() error) error {
|
|
if sd.postLoadSem == nil {
|
|
return fn()
|
|
}
|
|
|
|
start := time.Now()
|
|
if err := sd.postLoadSem.Acquire(ctx); err != nil {
|
|
return err
|
|
}
|
|
defer sd.postLoadSem.Release()
|
|
mlog.Debug(ctx, "delegator acquired post-load slot",
|
|
mlog.Duration("wait", time.Since(start)),
|
|
mlog.Int("capacity", sd.postLoadSem.Cap()),
|
|
mlog.Int("current", sd.postLoadSem.Current()))
|
|
|
|
return fn()
|
|
}
|
|
|
|
func (sd *shardDelegator) addDistributionIfSchemaBarrierOK(schemaBarrierTs uint64, entries ...SegmentEntry) error {
|
|
sd.schemaChangeMutex.RLock()
|
|
defer sd.schemaChangeMutex.RUnlock()
|
|
if schemaBarrierTs < sd.schemaBarrierTs {
|
|
return merr.WrapErrServiceInternal("schema barrier changed")
|
|
}
|
|
|
|
// alter distribution
|
|
sd.distribution.AddDistributions(entries...)
|
|
return nil
|
|
}
|
|
|
|
// LoadGrowing load growing segments locally.
|
|
func (sd *shardDelegator) LoadL0(ctx context.Context, infos []*querypb.SegmentLoadInfo, version int64) error {
|
|
log := sd.getLogger(ctx)
|
|
|
|
segmentIDs := lo.Map(infos, func(info *querypb.SegmentLoadInfo, _ int) int64 { return info.GetSegmentID() })
|
|
log.Info(ctx, "loading l0 segments...", mlog.Int64s("segmentIDs", segmentIDs))
|
|
|
|
loaded := make([]segments.Segment, 0)
|
|
if sd.l0ForwardPolicy == L0ForwardPolicyRemoteLoad {
|
|
for _, info := range infos {
|
|
l0Seg, err := segments.NewL0Segment(sd.collection, segments.SegmentTypeSealed, version, info)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
loaded = append(loaded, l0Seg)
|
|
}
|
|
} else {
|
|
var err error
|
|
loaded, err = sd.loader.Load(ctx, sd.collectionID, segments.SegmentTypeSealed, version, infos...)
|
|
if err != nil {
|
|
log.Warn(ctx, "failed to load l0 segment", mlog.Err(err))
|
|
return err
|
|
}
|
|
}
|
|
|
|
segmentIDs = lo.Map(loaded, func(segment segments.Segment, _ int) int64 { return segment.ID() })
|
|
log.Info(ctx, "load l0 segments done", mlog.Int64s("segmentIDs", segmentIDs))
|
|
|
|
sd.deleteBuffer.RegisterL0(loaded...)
|
|
// register l0 segment
|
|
sd.RefreshLevel0DeletionStats()
|
|
return nil
|
|
}
|
|
|
|
func (sd *shardDelegator) rangeHitL0Deletions(partitionID int64, candidate pkoracle.Candidate, fn func(pk storage.PrimaryKey, ts uint64) error) error {
|
|
level0Segments := sd.deleteBuffer.ListL0()
|
|
|
|
if len(level0Segments) == 0 {
|
|
return nil
|
|
}
|
|
|
|
log := sd.getLogger(context.Background())
|
|
start := time.Now()
|
|
totalL0Rows := 0
|
|
totalForwardRows := int64(0)
|
|
processedL0Count := 0
|
|
|
|
for _, segment := range level0Segments {
|
|
segment := segment.(*segments.L0Segment)
|
|
if segment.Partition() == partitionID || segment.Partition() == common.AllPartitionsID {
|
|
segmentPks, segmentTss := segment.DeleteRecords()
|
|
totalL0Rows += len(segmentPks)
|
|
processedL0Count++
|
|
batchSize := paramtable.Get().CommonCfg.BloomFilterApplyBatchSize.GetAsInt()
|
|
|
|
for idx := 0; idx < len(segmentPks); idx += batchSize {
|
|
endIdx := idx + batchSize
|
|
if endIdx > len(segmentPks) {
|
|
endIdx = len(segmentPks)
|
|
}
|
|
|
|
if !candidate.PkCandidateExist() {
|
|
for i := idx; i < endIdx; i++ {
|
|
totalForwardRows += 1
|
|
if err := fn(segmentPks[i], segmentTss[i]); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
continue
|
|
}
|
|
|
|
lc := storage.NewBatchLocationsCache(segmentPks[idx:endIdx])
|
|
hits := candidate.BatchPkExist(lc)
|
|
for i, hit := range hits {
|
|
if !hit {
|
|
continue
|
|
}
|
|
totalForwardRows += 1
|
|
if err := fn(segmentPks[idx+i], segmentTss[idx+i]); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
log.Info(context.TODO(), "forward delete from L0 segments to worker",
|
|
mlog.Int64("targetSegmentID", candidate.ID()),
|
|
mlog.String("channel", sd.vchannelName),
|
|
mlog.Bool("broadcast", !candidate.PkCandidateExist()),
|
|
mlog.Int("l0SegmentCount", processedL0Count),
|
|
mlog.Int("totalDeleteRowsInL0", totalL0Rows),
|
|
mlog.Int64("totalForwardRows", totalForwardRows),
|
|
mlog.Int64("totalCost", time.Since(start).Milliseconds()),
|
|
)
|
|
|
|
return nil
|
|
}
|
|
|
|
func (sd *shardDelegator) GetLevel0Deletions(partitionID int64, candidate pkoracle.Candidate) (storage.PrimaryKeys, []storage.Timestamp) {
|
|
deltaData := storage.NewDeltaData(0)
|
|
|
|
sd.rangeHitL0Deletions(partitionID, candidate, func(pk storage.PrimaryKey, ts uint64) error {
|
|
deltaData.Append(pk, ts)
|
|
return nil
|
|
})
|
|
|
|
return deltaData.DeletePks(), deltaData.DeleteTimestamps()
|
|
}
|
|
|
|
func (sd *shardDelegator) StreamForwardLevel0Deletions(bufferedForwarder *BufferForwarder, partitionID int64, candidate pkoracle.Candidate) error {
|
|
err := sd.rangeHitL0Deletions(partitionID, candidate, func(pk storage.PrimaryKey, ts uint64) error {
|
|
return bufferedForwarder.Buffer(pk, ts)
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return bufferedForwarder.Flush()
|
|
}
|
|
|
|
func (sd *shardDelegator) RefreshLevel0DeletionStats() {
|
|
level0Segments := sd.deleteBuffer.ListL0()
|
|
totalSize := int64(0)
|
|
for _, segment := range level0Segments {
|
|
segment := segment.(*segments.L0Segment)
|
|
pks, tss := segment.DeleteRecords()
|
|
totalSize += lo.SumBy(pks, func(pk storage.PrimaryKey) int64 { return pk.Size() }) + int64(len(tss)*8)
|
|
}
|
|
|
|
metrics.QueryNodeNumSegments.WithLabelValues(
|
|
paramtable.GetStringNodeID(),
|
|
fmt.Sprint(sd.Collection()),
|
|
commonpb.SegmentState_Sealed.String(),
|
|
datapb.SegmentLevel_L0.String(),
|
|
).Set(float64(len(level0Segments)))
|
|
|
|
metrics.QueryNodeLevelZeroSize.WithLabelValues(
|
|
paramtable.GetStringNodeID(),
|
|
fmt.Sprint(sd.collectionID),
|
|
sd.vchannelName,
|
|
).Set(float64(totalSize))
|
|
}
|
|
|
|
// processDeleteRecords performs BF checks on delete buffer records and forwards matching deletes
|
|
// via the buffered forwarder. Does NOT require any lock to be held.
|
|
// When candidate has no stats (PkCandidateExist() == false), all deletes are forwarded (broadcast mode).
|
|
// Returns the number of timestamp-hit and bloom-filter-hit rows.
|
|
func (sd *shardDelegator) processDeleteRecords(
|
|
candidate *pkoracle.BloomFilterSet,
|
|
records []*deletebuffer.Item,
|
|
forwarder *BufferForwarder,
|
|
) (tsHit, bfHit int64, err error) {
|
|
for _, entry := range records {
|
|
for _, record := range entry.Data {
|
|
tsHit += int64(len(record.DeleteData.Pks))
|
|
if record.PartitionID != common.AllPartitionsID && candidate.Partition() != record.PartitionID {
|
|
continue
|
|
}
|
|
pks := record.DeleteData.Pks
|
|
batchSize := paramtable.Get().CommonCfg.BloomFilterApplyBatchSize.GetAsInt()
|
|
for idx := 0; idx < len(pks); idx += batchSize {
|
|
endIdx := idx + batchSize
|
|
if endIdx > len(pks) {
|
|
endIdx = len(pks)
|
|
}
|
|
|
|
// When BF not initialized (bloom filter disabled), forward all deletes
|
|
if !candidate.PkCandidateExist() {
|
|
for i := idx; i < endIdx; i++ {
|
|
bfHit++
|
|
if err = forwarder.Buffer(pks[i], record.DeleteData.Tss[i]); err != nil {
|
|
return tsHit, bfHit, err
|
|
}
|
|
}
|
|
continue
|
|
}
|
|
|
|
lc := storage.NewBatchLocationsCache(pks[idx:endIdx])
|
|
hits := candidate.BatchPkExist(lc)
|
|
for i, hit := range hits {
|
|
if hit {
|
|
bfHit++
|
|
if err = forwarder.Buffer(pks[idx+i], record.DeleteData.Tss[idx+i]); err != nil {
|
|
return tsHit, bfHit, err
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return tsHit, bfHit, nil
|
|
}
|
|
|
|
// segDeleteSnapshot holds the snapshotted delete buffer entries for a segment,
|
|
// captured under RLock in Phase 1 of loadStreamDelete.
|
|
type segDeleteSnapshot struct {
|
|
records []*deletebuffer.Item // copied slice of delete buffer entries
|
|
snapshotMaxTs uint64 // max Item.Ts in snapshot, used for timestamp-based catch-up
|
|
}
|
|
|
|
func (sd *shardDelegator) loadStreamDelete(ctx context.Context,
|
|
candidates []*pkoracle.BloomFilterSet,
|
|
infos []*querypb.SegmentLoadInfo,
|
|
req *querypb.LoadSegmentsRequest,
|
|
targetNodeID int64,
|
|
worker cluster.Worker,
|
|
entries []SegmentEntry,
|
|
schemaBarrierTs uint64,
|
|
) error {
|
|
log := sd.getLogger(ctx)
|
|
|
|
idCandidates := lo.SliceToMap(candidates, func(candidate *pkoracle.BloomFilterSet) (int64, *pkoracle.BloomFilterSet) {
|
|
return candidate.ID(), candidate
|
|
})
|
|
|
|
// Phase 0: Forward L0 deletions (no lock needed, unchanged)
|
|
for _, info := range infos {
|
|
candidate := idCandidates[info.GetSegmentID()]
|
|
err := sd.forwardL0Deletion(ctx, info, req, candidate, targetNodeID, worker)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
// === Phase 1: Snapshot delete buffer entries under RLock (fast — microseconds) ===
|
|
sd.deleteMut.RLock()
|
|
snapshots := make([]segDeleteSnapshot, len(infos))
|
|
for i, info := range infos {
|
|
records := sd.deleteBuffer.ListAfter(segmentEffectiveTs(info))
|
|
// Copy the slice to safely use outside lock scope.
|
|
// ListAfter returns a new slice from doubleCacheBuffer, but we copy to
|
|
// ensure no dependency on internal buffer state that may change after unlock.
|
|
copied := make([]*deletebuffer.Item, len(records))
|
|
copy(copied, records)
|
|
var maxTs uint64
|
|
if len(records) > 0 {
|
|
maxTs = records[len(records)-1].Ts
|
|
}
|
|
snapshots[i] = segDeleteSnapshot{
|
|
records: copied,
|
|
snapshotMaxTs: maxTs,
|
|
}
|
|
}
|
|
sd.deleteMut.RUnlock()
|
|
// RLock released — WAL pipeline (ProcessDelete) is now unblocked
|
|
|
|
// Create one forwarder per segment, shared across Phase 2 and Phase 3, flushed once at the end.
|
|
forwarders := make([]*BufferForwarder, len(infos))
|
|
for i, info := range infos {
|
|
candidate := idCandidates[info.GetSegmentID()]
|
|
deleteScope := querypb.DataScope_All
|
|
switch candidate.Type() {
|
|
case commonpb.SegmentState_Sealed:
|
|
deleteScope = querypb.DataScope_Historical
|
|
case commonpb.SegmentState_Growing:
|
|
deleteScope = querypb.DataScope_Streaming
|
|
}
|
|
forwarders[i] = NewBufferedForwarder(paramtable.Get().QueryNodeCfg.ForwardBatchSize.GetAsInt64(),
|
|
deleteViaWorker(ctx, worker, targetNodeID, info, deleteScope))
|
|
}
|
|
|
|
// === Phase 2: Process snapshot WITHOUT lock (expensive — seconds) ===
|
|
for i, info := range infos {
|
|
candidate := idCandidates[info.GetSegmentID()]
|
|
start := time.Now()
|
|
tsHit, bfHit, err := sd.processDeleteRecords(candidate, snapshots[i].records, forwarders[i])
|
|
if err != nil {
|
|
return err
|
|
}
|
|
log.Info(ctx, "forward delete to worker (phase 2: snapshot)...",
|
|
mlog.String("channel", info.InsertChannel),
|
|
mlog.FieldSegmentID(info.GetSegmentID()),
|
|
mlog.Time("startPosition", tsoutil.PhysicalTime(info.GetStartPosition().GetTimestamp())),
|
|
mlog.Int64("tsHitDeleteRowNum", tsHit),
|
|
mlog.Int64("bfHitDeleteRowNum", bfHit),
|
|
mlog.Int64("bfCost", time.Since(start).Milliseconds()),
|
|
)
|
|
}
|
|
|
|
// === Phase 3: Catch-up new entries + flush + add distribution under RLock (fast — milliseconds) ===
|
|
sd.deleteMut.RLock()
|
|
defer sd.deleteMut.RUnlock()
|
|
|
|
for i, info := range infos {
|
|
candidate := idCandidates[info.GetSegmentID()]
|
|
|
|
// Use timestamp-based catch-up: fetch records added after the snapshot's max timestamp.
|
|
// This is robust against delete buffer eviction (Put → evict discards old tail during Phase 2).
|
|
// Index-based approach (allRecords[snapshotLen:]) would panic or miss data if eviction occurs.
|
|
// Item.Ts comes from WAL TSO, monotonically increasing and unique per ProcessDelete call,
|
|
// so ListAfter(snapshotMaxTs + 1) precisely captures only new records.
|
|
catchUpTs := segmentEffectiveTs(info)
|
|
if snapshots[i].snapshotMaxTs > 0 {
|
|
catchUpTs = snapshots[i].snapshotMaxTs + 1
|
|
}
|
|
newRecords := sd.deleteBuffer.ListAfter(catchUpTs)
|
|
if len(newRecords) > 0 {
|
|
start := time.Now()
|
|
tsHit, bfHit, err := sd.processDeleteRecords(candidate, newRecords, forwarders[i])
|
|
if err != nil {
|
|
return err
|
|
}
|
|
log.Info(ctx, "forward delete to worker (phase 3: catch-up)...",
|
|
mlog.String("channel", info.InsertChannel),
|
|
mlog.FieldSegmentID(info.GetSegmentID()),
|
|
mlog.Int64("tsHitDeleteRowNum", tsHit),
|
|
mlog.Int64("bfHitDeleteRowNum", bfHit),
|
|
mlog.Int64("bfCost", time.Since(start).Milliseconds()),
|
|
)
|
|
}
|
|
|
|
// Flush once per segment after both phases are done
|
|
if err := forwarders[i].Flush(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
// Atomically add to distribution while still holding RLock.
|
|
// This guarantees no ProcessDelete can run between catch-up and distribution update,
|
|
// so there is no gap between "deletes applied" and "segment visible".
|
|
if err := sd.addDistributionIfSchemaBarrierOK(schemaBarrierTs, entries...); err != nil {
|
|
return err
|
|
}
|
|
log.Info(ctx, "load stream delete done")
|
|
return nil
|
|
}
|
|
|
|
// ReleaseSegments releases segments local or remotely depending on the target node.
|
|
func (sd *shardDelegator) ReleaseSegments(ctx context.Context, req *querypb.ReleaseSegmentsRequest, force bool) error {
|
|
log := sd.getLogger(ctx)
|
|
|
|
targetNodeID := req.GetNodeID()
|
|
level0Segments := typeutil.NewSet(lo.Map(sd.deleteBuffer.ListL0(), func(segment segments.Segment, _ int) int64 {
|
|
return segment.ID()
|
|
})...)
|
|
hasLevel0 := false
|
|
for _, segmentID := range req.GetSegmentIDs() {
|
|
hasLevel0 = level0Segments.Contain(segmentID)
|
|
if hasLevel0 {
|
|
return merr.WrapErrServiceInternal("release L0 segment is not supported, l0 segment should only be released by unSubChannel/SyncDataDistribution")
|
|
}
|
|
}
|
|
|
|
// add common log fields
|
|
log = log.With(
|
|
mlog.Int64s("segmentIDs", req.GetSegmentIDs()),
|
|
mlog.FieldNodeID(req.GetNodeID()),
|
|
mlog.String("scope", req.GetScope().String()),
|
|
mlog.Bool("force", force))
|
|
|
|
log.Info(ctx, "delegator start to release segments")
|
|
// alter distribution first
|
|
var sealed, growing []SegmentEntry
|
|
convertSealed := func(segmentID int64, _ int) SegmentEntry {
|
|
return SegmentEntry{
|
|
SegmentID: segmentID,
|
|
NodeID: targetNodeID,
|
|
}
|
|
}
|
|
convertGrowing := func(segmentID int64, _ int) SegmentEntry {
|
|
return SegmentEntry{
|
|
SegmentID: segmentID,
|
|
}
|
|
}
|
|
switch req.GetScope() {
|
|
case querypb.DataScope_All:
|
|
sealed = lo.Map(req.GetSegmentIDs(), convertSealed)
|
|
growing = lo.Map(req.GetSegmentIDs(), convertGrowing)
|
|
case querypb.DataScope_Streaming:
|
|
growing = lo.Map(req.GetSegmentIDs(), convertGrowing)
|
|
case querypb.DataScope_Historical:
|
|
sealed = lo.Map(req.GetSegmentIDs(), convertSealed)
|
|
}
|
|
|
|
signal := sd.distribution.RemoveDistributions(sealed, growing)
|
|
// wait cleared signal
|
|
<-signal
|
|
|
|
if len(growing) > 0 {
|
|
sd.growingSegmentLock.Lock()
|
|
}
|
|
// when we try to release a segment, add it to pipeline's exclude list first
|
|
// in case of consumed it's growing segment again
|
|
droppedInfos := lo.SliceToMap(req.GetSegmentIDs(), func(id int64) (int64, uint64) {
|
|
if req.GetCheckpoint() == nil {
|
|
return id, typeutil.MaxTimestamp
|
|
}
|
|
|
|
return id, req.GetCheckpoint().GetTimestamp()
|
|
})
|
|
sd.AddExcludedSegments(droppedInfos)
|
|
|
|
var releaseErr error
|
|
if !force {
|
|
worker, err := sd.workerManager.GetWorker(ctx, targetNodeID)
|
|
if err != nil {
|
|
log.Warn(ctx, "delegator failed to find worker", mlog.Err(err))
|
|
releaseErr = err
|
|
}
|
|
req.Base.TargetID = targetNodeID
|
|
err = worker.ReleaseSegments(ctx, req)
|
|
if err != nil {
|
|
log.Warn(ctx, "worker failed to release segments", mlog.Err(err))
|
|
releaseErr = err
|
|
}
|
|
}
|
|
if len(growing) > 0 {
|
|
sd.growingSegmentLock.Unlock()
|
|
}
|
|
|
|
if releaseErr != nil {
|
|
return releaseErr
|
|
}
|
|
if len(growing) > 0 && sd.growingSourceProvider != nil {
|
|
for _, entry := range growing {
|
|
sd.growingSourceProvider.ClearReleasePrepared(entry.SegmentID)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (sd *shardDelegator) SyncTargetVersion(action *querypb.SyncAction, partitions []int64) {
|
|
sd.distribution.SyncTargetVersion(action, partitions)
|
|
// clean delete buffer after distribution becomes serviceable
|
|
if sd.distribution.queryView.Serviceable() {
|
|
checkpoint := action.GetCheckpoint()
|
|
deleteSeekPos := action.GetDeleteCP()
|
|
if deleteSeekPos == nil {
|
|
// for compatible with 2.4, we use checkpoint as deleteCP when deleteCP is nil
|
|
deleteSeekPos = checkpoint
|
|
mlog.Info(context.TODO(), "use checkpoint as deleteCP",
|
|
mlog.String("channelName", sd.vchannelName),
|
|
mlog.Time("deleteSeekPos", tsoutil.PhysicalTime(action.GetCheckpoint().GetTimestamp())))
|
|
}
|
|
|
|
start := time.Now()
|
|
sizeBeforeClean, _ := sd.deleteBuffer.Size()
|
|
l0NumBeforeClean := len(sd.deleteBuffer.ListL0())
|
|
sd.deleteBuffer.UnRegister(deleteSeekPos.GetTimestamp())
|
|
sizeAfterClean, _ := sd.deleteBuffer.Size()
|
|
l0NumAfterClean := len(sd.deleteBuffer.ListL0())
|
|
|
|
if sizeAfterClean < sizeBeforeClean || l0NumAfterClean < l0NumBeforeClean {
|
|
mlog.Info(context.TODO(), "clean delete buffer",
|
|
mlog.String("channel", sd.vchannelName),
|
|
mlog.Time("deleteSeekPos", tsoutil.PhysicalTime(deleteSeekPos.GetTimestamp())),
|
|
mlog.Time("channelCP", tsoutil.PhysicalTime(checkpoint.GetTimestamp())),
|
|
mlog.Int64("sizeBeforeClean", sizeBeforeClean),
|
|
mlog.Int64("sizeAfterClean", sizeAfterClean),
|
|
mlog.Int("l0NumBeforeClean", l0NumBeforeClean),
|
|
mlog.Int("l0NumAfterClean", l0NumAfterClean),
|
|
mlog.Duration("cost", time.Since(start)),
|
|
)
|
|
}
|
|
sd.RefreshLevel0DeletionStats()
|
|
}
|
|
}
|
|
|
|
func (sd *shardDelegator) GetChannelQueryView() *channelQueryView {
|
|
return sd.distribution.GetQueryView()
|
|
}
|
|
|
|
func (sd *shardDelegator) AddExcludedSegments(excludeInfo map[int64]uint64) {
|
|
sd.excludedSegments.Insert(excludeInfo)
|
|
}
|
|
|
|
func (sd *shardDelegator) VerifyExcludedSegments(segmentID int64, ts uint64) bool {
|
|
return sd.excludedSegments.Verify(segmentID, ts)
|
|
}
|
|
|
|
func (sd *shardDelegator) TryCleanExcludedSegments(ts uint64) {
|
|
if sd.excludedSegments.ShouldClean() {
|
|
sd.excludedSegments.CleanInvalid(ts)
|
|
}
|
|
}
|
|
|
|
func (sd *shardDelegator) buildBM25IDF(ctx context.Context, req *internalpb.SearchRequest) (float64, error) {
|
|
idfOracle := sd.getIDFOracle()
|
|
if idfOracle == nil {
|
|
return 0, merr.WrapErrServiceInternal("bm25 oracle is not initialized")
|
|
}
|
|
|
|
pb := &commonpb.PlaceholderGroup{}
|
|
if err := proto.Unmarshal(req.GetPlaceholderGroup(), pb); err != nil {
|
|
return 0, merr.WrapErrParameterInvalidMsg("failed to unmarshal BM25 IDF placeholder group: %v", err)
|
|
}
|
|
|
|
if len(pb.Placeholders) != 1 || len(pb.Placeholders[0].Values) == 0 {
|
|
return 0, merr.WrapErrParameterInvalidMsg("please provide varchar/text for BM25 Function based search")
|
|
}
|
|
|
|
holder := pb.Placeholders[0]
|
|
if holder.Type != commonpb.PlaceholderType_VarChar {
|
|
return 0, merr.WrapErrParameterInvalidMsg("please provide varchar/text for BM25 Function based search, got %s", holder.Type.String())
|
|
}
|
|
|
|
texts := funcutil.GetVarCharFromPlaceholder(holder)
|
|
var tfArray *schemapb.SparseFloatArray
|
|
schemaVersion := sd.collection.Schema().GetVersion()
|
|
ok, err := function.RunWithRunner(ctx, sd.collectionID, schemaVersion, req.GetFieldId(), func(functionType schemapb.FunctionType, functionRunner function.FunctionRunner) error {
|
|
if functionType != schemapb.FunctionType_BM25 {
|
|
return merr.WrapErrServiceInternalMsg("functionRunner not found for field: %d", req.GetFieldId())
|
|
}
|
|
|
|
datas := []any{texts}
|
|
if len(functionRunner.GetInputFields()) == 2 {
|
|
analyzerName := "default"
|
|
if name := req.GetAnalyzerName(); name != "" {
|
|
// use user provided analyzer name
|
|
analyzerName = name
|
|
}
|
|
|
|
analyzers := make([]string, len(texts))
|
|
for i := range texts {
|
|
analyzers[i] = analyzerName
|
|
}
|
|
datas = append(datas, analyzers)
|
|
}
|
|
|
|
// get search text term frequency
|
|
output, err := functionRunner.BatchRun(datas...)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(output) == 0 {
|
|
return merr.WrapErrFunctionFailedMsg("BM25 embedding failed: runner returned empty output")
|
|
}
|
|
|
|
var ok bool
|
|
tfArray, ok = output[0].(*schemapb.SparseFloatArray)
|
|
if !ok {
|
|
return merr.WrapErrFunctionFailedMsg("functionRunner return unknown data")
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if !ok {
|
|
// internal invariant: runners are populated with the schema, never by
|
|
// the request — classified system, keeps cross-replica failover
|
|
return 0, merr.WrapErrServiceInternalMsg("functionRunner not found for field: %d", req.GetFieldId())
|
|
}
|
|
|
|
idfSparseVector, avgdl, err := idfOracle.BuildIDF(req.GetFieldId(), tfArray)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
if avgdl <= 0 {
|
|
return 0, nil
|
|
}
|
|
|
|
for _, idf := range idfSparseVector {
|
|
metrics.QueryNodeSearchFTSNumTokens.WithLabelValues(paramtable.GetStringNodeID(), fmt.Sprint(sd.collectionID), fmt.Sprint(req.GetFieldId())).Observe(float64(typeutil.SparseFloatRowElementCount(idf)))
|
|
}
|
|
|
|
err = SetBM25Params(req, avgdl)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
req.PlaceholderGroup = funcutil.SparseVectorDataToPlaceholderGroupBytes(idfSparseVector)
|
|
return avgdl, nil
|
|
}
|
|
|
|
func (sd *shardDelegator) parseMinHash(ctx context.Context, req *internalpb.SearchRequest) error {
|
|
pb := &commonpb.PlaceholderGroup{}
|
|
if err := proto.Unmarshal(req.GetPlaceholderGroup(), pb); err != nil {
|
|
return merr.WrapErrParameterInvalidMsg("failed to unmarshal MinHash placeholder group: %v", err)
|
|
}
|
|
|
|
if len(pb.Placeholders) != 1 || len(pb.Placeholders[0].Values) == 0 {
|
|
return merr.WrapErrParameterInvalidMsg("please provide varchar/text for MinHash Function based search")
|
|
}
|
|
|
|
holder := pb.Placeholders[0]
|
|
if holder.Type != commonpb.PlaceholderType_VarChar {
|
|
return merr.WrapErrParameterInvalidMsg("please provide varchar/text for MinHash Function based search, got %s", holder.Type.String())
|
|
}
|
|
|
|
texts := funcutil.GetVarCharFromPlaceholder(holder)
|
|
var fieldData *schemapb.FieldData
|
|
schemaVersion := sd.collection.Schema().GetVersion()
|
|
ok, err := function.RunWithRunner(ctx, sd.collectionID, schemaVersion, req.GetFieldId(), func(functionType schemapb.FunctionType, functionRunner function.FunctionRunner) error {
|
|
if functionType != schemapb.FunctionType_MinHash {
|
|
return merr.WrapErrServiceInternalMsg("functionRunner not found for field: %d", req.GetFieldId())
|
|
}
|
|
|
|
output, err := functionRunner.BatchRun(texts)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(output) == 0 {
|
|
return merr.WrapErrFunctionFailedMsg("MinHash embedding failed: runner returned empty output")
|
|
}
|
|
|
|
var ok bool
|
|
fieldData, ok = output[0].(*schemapb.FieldData)
|
|
if !ok {
|
|
return merr.WrapErrFunctionFailedMsg("MinHash embedding failed: MinHash functionRunner return unknown data")
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !ok {
|
|
return merr.WrapErrServiceInternalMsg("functionRunner not found for field: %d", req.GetFieldId())
|
|
}
|
|
|
|
vectorField := fieldData.GetVectors()
|
|
if vectorField == nil {
|
|
return merr.WrapErrFunctionFailedMsg("MinHash embedding failed: output is not a vector field")
|
|
}
|
|
|
|
binaryVector := vectorField.GetBinaryVector()
|
|
if binaryVector == nil {
|
|
return merr.WrapErrFunctionFailedMsg("MinHash embedding failed: output is not a binary vector")
|
|
}
|
|
|
|
req.PlaceholderGroup, err = funcutil.FieldDataToPlaceholderGroupBytes(fieldData)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (sd *shardDelegator) DropIndex(ctx context.Context, req *querypb.DropIndexRequest) error {
|
|
workers := sd.workerManager.GetAllWorkers()
|
|
for _, worker := range workers {
|
|
if err := worker.DropIndex(ctx, req); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (sd *shardDelegator) GetHighlight(ctx context.Context, req *querypb.GetHighlightRequest) ([]*querypb.HighlightResult, error) {
|
|
result := []*querypb.HighlightResult{}
|
|
for _, task := range req.GetTasks() {
|
|
if len(task.GetTexts()) != int(task.GetSearchTextNum()+task.GetCorpusTextNum())+len(task.GetQueries()) {
|
|
return nil, merr.WrapErrServiceInternalMsg("package highlight texts error, num of texts not equal the expected num %d:%d", len(task.GetTexts()), int(task.GetSearchTextNum()+task.GetCorpusTextNum())+len(task.GetQueries()))
|
|
}
|
|
topks := req.GetTopks()
|
|
var results [][]*milvuspb.AnalyzerToken
|
|
var analyzeErr error
|
|
ok, err := sd.runWithAnalyzer(ctx, task.GetFieldId(), func(analyzer function.Analyzer) error {
|
|
if len(analyzer.GetInputFields()) == 1 {
|
|
results, analyzeErr = analyzer.BatchAnalyze(true, false, task.GetTexts())
|
|
return analyzeErr
|
|
}
|
|
if len(analyzer.GetInputFields()) == 2 {
|
|
analyzerNames, err := normalizeHighlightAnalyzerNames(task.GetAnalyzerNames(), len(task.GetTexts()))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
results, analyzeErr = analyzer.BatchAnalyze(true, false, task.GetTexts(), analyzerNames)
|
|
return analyzeErr
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !ok {
|
|
return nil, merr.WrapErrParameterInvalidMsg("get highlight failed, the highlight field not found, %s:%d", task.GetFieldName(), task.GetFieldId())
|
|
}
|
|
|
|
// analyze result of search text
|
|
searchResults := results[0:task.SearchTextNum]
|
|
// analyze result of query text
|
|
queryResults := results[task.SearchTextNum : task.SearchTextNum+int64(len(task.Queries))]
|
|
// analyze result of corpus text
|
|
corpusStartOffset := int(task.SearchTextNum) + len(task.Queries)
|
|
corpusResults := results[corpusStartOffset:]
|
|
|
|
// query for all corpus texts
|
|
// only support text match now
|
|
// build match set for all analyze result of query text
|
|
// TODO: support more query types
|
|
queryTokenSet := typeutil.NewSet[string]()
|
|
for _, tokens := range queryResults {
|
|
for _, token := range tokens {
|
|
queryTokenSet.Insert(token.GetToken())
|
|
}
|
|
}
|
|
|
|
corpusIdx := 0
|
|
for i := range len(topks) {
|
|
tokenSet := typeutil.NewSet[string]()
|
|
if len(searchResults) > i {
|
|
for _, token := range searchResults[i] {
|
|
tokenSet.Insert(token.GetToken())
|
|
}
|
|
}
|
|
|
|
for j := 0; j < int(topks[i]); j++ {
|
|
spans := SpanList{}
|
|
for _, token := range corpusResults[corpusIdx] {
|
|
if tokenSet.Contain(token.GetToken()) || queryTokenSet.Contain(token.GetToken()) {
|
|
spans = append(spans, Span{token.GetStartOffset(), token.GetEndOffset()})
|
|
}
|
|
}
|
|
spans = mergeOffsets(spans)
|
|
|
|
// Convert byte offsets from analyzer to rune (character) offsets
|
|
corpusText := task.Texts[corpusStartOffset+corpusIdx]
|
|
err := bytesOffsetToRuneOffset(corpusText, spans)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
frags := fetchFragmentsFromOffsets(corpusText, spans,
|
|
task.GetOptions().GetFragmentOffset(),
|
|
task.GetOptions().GetFragmentSize(),
|
|
task.GetOptions().GetNumOfFragments())
|
|
result = append(result, &querypb.HighlightResult{Fragments: frags})
|
|
corpusIdx++
|
|
}
|
|
}
|
|
}
|
|
|
|
return result, nil
|
|
}
|