Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix: Check resource when loading deltalogs #37195

Merged
merged 2 commits into from
Oct 29, 2024
Merged

fix: Check resource when loading deltalogs #37195

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ddc3ba2
fix: Check resource when loading deltalogs
congqixia Oct 28, 2024
16d3ca7
Fix collection not filled in delta load info
congqixia Oct 28, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 24 additions & 2 deletions internal/querynodev2/segments/segment_loader.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,8 @@
// NOTE: make sure the ref count of the corresponding collection will never go down to 0 during this
Load(ctx context.Context, collectionID int64, segmentType SegmentType, version int64, segments ...*querypb.SegmentLoadInfo) ([]Segment, error)

// LoadDeltaLogs load deltalog and write delta data into provided segment.
// it also executes resource protection logic in case of OOM.
LoadDeltaLogs(ctx context.Context, segment Segment, deltaLogs []*datapb.FieldBinlog) error

// LoadBloomFilterSet loads needed statslog for RemoteSegment.
Expand Down Expand Up @@ -334,7 +336,7 @@
}
}
}
if err = loader.LoadDeltaLogs(ctx, segment, loadInfo.GetDeltalogs()); err != nil {
if err = loader.loadDeltalogs(ctx, segment, loadInfo.GetDeltalogs()); err != nil {
return errors.Wrap(err, "At LoadDeltaLogs")
}

Expand Down Expand Up @@ -1158,7 +1160,9 @@
return nil
}

func (loader *segmentLoader) LoadDeltaLogs(ctx context.Context, segment Segment, deltaLogs []*datapb.FieldBinlog) error {
// loadDeltalogs performs the internal actions of `LoadDeltaLogs`
// this function does not perform resource check and is meant be used among other load APIs.
func (loader *segmentLoader) loadDeltalogs(ctx context.Context, segment Segment, deltaLogs []*datapb.FieldBinlog) error {
ctx, sp := otel.Tracer(typeutil.QueryNodeRole).Start(ctx, fmt.Sprintf("LoadDeltalogs-%d", segment.ID()))
defer sp.End()
log := log.Ctx(ctx).With(
Expand Down Expand Up @@ -1254,6 +1258,24 @@
return nil
}

// LoadDeltaLogs load deltalog and write delta data into provided segment.
// it also executes resource protection logic in case of OOM.
func (loader *segmentLoader) LoadDeltaLogs(ctx context.Context, segment Segment, deltaLogs []*datapb.FieldBinlog) error {
loadInfo := &querypb.SegmentLoadInfo{
SegmentID: segment.ID(),
CollectionID: segment.Collection(),
Deltalogs: deltaLogs,
}
// Check memory & storage limit
requestResourceResult, err := loader.requestResource(ctx, loadInfo)
if err != nil {
log.Warn("request resource failed", zap.Error(err))
return err
}

Check warning on line 1274 in internal/querynodev2/segments/segment_loader.go

View check run for this annotation

Codecov / codecov/patch

internal/querynodev2/segments/segment_loader.go#L1272-L1274 

Added lines #L1272 - L1274 were not covered by tests
defer loader.freeRequest(requestResourceResult.Resource)
return loader.loadDeltalogs(ctx, segment, deltaLogs)
}

func (loader *segmentLoader) patchEntryNumber(ctx context.Context, segment *LocalSegment, loadInfo *querypb.SegmentLoadInfo) error {
var needReset bool

Expand Down
Loading