Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
Commits
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
44 changes: 44 additions & 0 deletions index/scorch/mergeplan/merge_plan.go
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,22 @@ type MergePlanOptions struct {
// impact merge selection.
ReclaimDeletesWeight float64

// DeletesPctAllowed makes a segment a merge candidate on its deleted
// fraction alone, whatever its size.
//
// Without it a large segment can hold tombstones indefinitely. Eligibility
// is liveSize < MaxSegmentSize/2, and liveSize already excludes deletes, so
// deleting does pull a segment back towards eligibility -- but only slowly.
// A merge roster is capped at MaxSegmentSize, so segments freeze somewhere
// between MaxSegmentSize/2 and MaxSegmentSize, and from there need half to
// three quarters of their documents deleted before they qualify again.
// Until that point ReclaimDeletesWeight cannot help either, because scoring
// only ever runs over the eligible set.
//
// Set it too low and the rewrite costs more than it reclaims; below roughly
// a quarter is rarely worth it. Zero disables the behaviour entirely.
DeletesPctAllowed float64

// Optional, defaults to mergeplan.CalcBudget().
CalcBudget func(totalSize int64, firstTierSize int64,
o *MergePlanOptions) (budgetNumSegments int)
Expand Down Expand Up @@ -165,6 +181,7 @@ var DefaultMergePlanOptions = MergePlanOptions{
SegmentsPerMergeTask: 10,
FloorSegmentSize: 2000,
ReclaimDeletesWeight: 2.0,
DeletesPctAllowed: 0.33,
}

// SingleSegmentMergePlanOptions helps in creating a
Expand Down Expand Up @@ -252,6 +269,7 @@ func plan(segmentsIn []Segment, o *MergePlanOptions) (*MergePlan, error) {
var minLiveFileSize int64 = math.MaxInt64

var eligibles []Segment
var deleteHeavy []Segment
var eligiblesLiveSize int64
var eligiblesLiveFileSize int64

Expand All @@ -275,6 +293,13 @@ func plan(segmentsIn []Segment, o *MergePlanOptions) (*MergePlan, error) {
if segment.HasVector() {
isEligible = isEligible && liveFileSize < o.MaxSegmentFileSize/2
}
// A segment too large to merge on size is still worth rewriting once
// enough of it is deleted; collect those separately, since one such
// segment can never exceed the budget and so would never be planned.
if !isEligible && o.DeletesPctAllowed > 0 &&
deletedFraction(segment) >= o.DeletesPctAllowed {
deleteHeavy = append(deleteHeavy, segment)
}

// Only small-enough segments are eligible.
if isEligible {
Expand Down Expand Up @@ -321,6 +346,12 @@ func plan(segmentsIn []Segment, o *MergePlanOptions) (*MergePlan, error) {
eligibles = removeSegments(eligibles, empties)
}

// Segments too large to be eligible on size, but deleted enough to be worth
// rewriting on their own.
for _, dh := range deleteHeavy {
rv.Tasks = append(rv.Tasks, &MergeTask{Segments: []Segment{dh}})
}

numMergeTasks := 0

// While we're over budget, keep looping, which might produce
Expand Down Expand Up @@ -392,6 +423,19 @@ func plan(segmentsIn []Segment, o *MergePlanOptions) (*MergePlan, error) {
return rv, nil
}

// deletedFraction is the share of a segment's documents that are deleted.
func deletedFraction(s Segment) float64 {
full := s.FullSize()
if full <= 0 {
return 0
}
live := s.LiveSize()
if live >= full {
return 0
}
return float64(full-live) / float64(full)
}

// Compute the number of segments that would be needed to cover the
// totalSize, by climbing up a logarithmically growing staircase of
// segment tiers.
Expand Down
170 changes: 170 additions & 0 deletions index/scorch/mergeplan/merge_plan_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -1125,3 +1125,173 @@ func TestPlanWithNoBestRoster(t *testing.T) {
t.Fatalf("final layout mismatch, got: %+v", spec.segments)
}
}

// A segment past the size-eligibility cut is never reconsidered however many
// of its documents are deleted: one segment cannot push len(eligibles) over
// the budget, so the roster loop never selects it, and ScoreSegments'
// ReclaimDeletesWeight only ever sees segments that already passed eligibility.
// DeletesPctAllowed is the escape.

// hasSingletonTaskFor reports whether the plan rewrites seg on its own, which
// is how a delete-heavy segment reaches the merger -- it can never push
// len(eligibles) over the budget, so the roster loop would never select it.
func hasSingletonTaskFor(p *MergePlan, seg Segment) bool {
if p == nil {
return false
}
for _, task := range p.Tasks {
if len(task.Segments) == 1 && task.Segments[0] == seg {
return true
}
}
return false
}

// A segment past the size cut is normally frozen for good. DeletesPctAllowed is
// the escape, but only within a window: a segment stays ineligible while
// liveSize >= MaxSegmentSize/2, so once enough is deleted it drops back under
// the cut and rejoins the ordinary path. At MyFullSize == MaxSegmentSize that
// window is roughly [threshold, 50%) deleted.
func TestPlanDeletesPctAllowed(t *testing.T) {
base := DefaultMergePlanOptions
full := base.MaxSegmentSize // the largest a roster can produce
cut := base.MaxSegmentSize / 2

cases := []struct {
name string
deletedPct float64
allowed float64
want bool
}{
{"pristine", 0.00, 0.33, false},
{"lightly deleted", 0.10, 0.33, false},
{"just under threshold", 0.32, 0.33, false},
{"exactly at threshold", 0.33, 0.33, true},
{"well over threshold", 0.45, 0.33, true},
{"deleted back under the size cut", 0.55, 0.33, false},
{"disabled by zero", 0.45, 0.00, false},
{"lower threshold catches less deleted", 0.10, 0.05, true},
}

for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
o := base
o.DeletesPctAllowed = tc.allowed

live := int64(float64(full) * (1 - tc.deletedPct))
fat := &segment{MyId: 1, MyFullSize: full, MyLiveSize: live,
MyFileSize: full * 120}

// enough ordinary segments that the planner has normal work to do
segs := []Segment{fat}
for i := 0; i < 12; i++ {
segs = append(segs, &segment{MyId: uint64(i + 2),
MyFullSize: 1000, MyLiveSize: 1000, MyFileSize: 100000})
}

p, err := plan(segs, &o)
if err != nil {
t.Fatal(err)
}

got := hasSingletonTaskFor(p, Segment(fat))
if got != tc.want {
t.Fatalf("deleted %.0f%%, DeletesPctAllowed %.2f, live %d vs cut %d: "+
"single-segment task = %v, want %v",
tc.deletedPct*100, tc.allowed, live, cut, got, tc.want)
}

// whichever way it went, the segment must never appear twice --
// once via the delete pass and again inside a roster
seen := 0
if p != nil {
for _, task := range p.Tasks {
for _, s := range task.Segments {
if s == Segment(fat) {
seen++
}
}
}
}
if seen > 1 {
t.Fatalf("segment planned %d times in one plan", seen)
}
})
}
}

// The delete pass must not disturb ordinary planning: small segments over
// budget should still be rostered in the same plan that rewrites a
// delete-heavy one.
func TestPlanDeletesPctAllowedLeavesNormalWorkIntact(t *testing.T) {
o := DefaultMergePlanOptions
o.DeletesPctAllowed = 0.33

full := o.MaxSegmentSize
fat := &segment{MyId: 1, MyFullSize: full,
MyLiveSize: int64(float64(full) * 0.6), MyFileSize: full * 120}

segs := []Segment{fat}
for i := 0; i < 40; i++ {
segs = append(segs, &segment{MyId: uint64(i + 2),
MyFullSize: 1000, MyLiveSize: 1000, MyFileSize: 100000})
}

p, err := plan(segs, &o)
if err != nil {
t.Fatal(err)
}
if p == nil {
t.Fatal("expected a plan: 40 small segments is well over budget")
}
if !hasSingletonTaskFor(p, Segment(fat)) {
t.Error("delete-heavy segment was not rewritten")
}

var rosters int
for _, task := range p.Tasks {
if len(task.Segments) > 1 {
rosters++
}
}
if rosters == 0 {
t.Errorf("no ordinary merge task planned alongside; got %d task(s)",
len(p.Tasks))
}
}

// An ineligible segment must not be counted towards the budget, or its presence
// would suppress merges of the small segments around it.
func TestPlanIneligibleSegmentDoesNotConsumeBudget(t *testing.T) {
o := DefaultMergePlanOptions
o.DeletesPctAllowed = 0 // isolate the budget question from the delete pass

small := func(n int) []Segment {
var rv []Segment
for i := 0; i < n; i++ {
rv = append(rv, &segment{MyId: uint64(i + 2),
MyFullSize: 1000, MyLiveSize: 1000, MyFileSize: 100000})
}
return rv
}

without, err := plan(small(40), &o)
if err != nil {
t.Fatal(err)
}
fat := &segment{MyId: 1, MyFullSize: o.MaxSegmentSize,
MyLiveSize: o.MaxSegmentSize - 1, MyFileSize: o.MaxSegmentSize * 120}
with, err := plan(append([]Segment{fat}, small(40)...), &o)
if err != nil {
t.Fatal(err)
}

if (without == nil) != (with == nil) {
t.Fatalf("plan presence changed by an ineligible segment: without=%v with=%v",
without != nil, with != nil)
}
if without != nil && len(without.Tasks) != len(with.Tasks) {
t.Errorf("ineligible segment changed task count: %d -> %d",
len(without.Tasks), len(with.Tasks))
}
}
Loading