5239711038

Committed 12 Jun 2023 05:16AM UTC coverage: 61.327% (+0.5%) from 60.801%

Build Type

push

Committed by

Commit Message

Merge 08a3e2d9a into 907dd6578

Run Details

411 of 411 new or added lines in 10 files covered. (100.0%)

8893 of 14501 relevant lines covered (61.33%)

449498.66 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.58

/level_handler.go

/*
 * Copyright 2017 Dgraph Labs, Inc. and Contributors
 *
 * 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.
 */

package badger

import (
        "fmt"
        "sort"
        "sync"

        "github.com/dgraph-io/badger/v4/table"
        "github.com/dgraph-io/badger/v4/y"
)

type levelHandler struct {
        // Guards tables, totalSize.
        sync.RWMutex

        // For level >= 1, tables are sorted by key ranges, which do not overlap.
        // For level 0, tables are sorted by time.
        // For level 0, newest table are at the back. Compact the oldest one first, which is at the front.
        tables         []*table.Table
        totalSize      int64
        totalStaleSize int64

        // The following are initialized once and const.
        level    int
        strLevel string
        db       *DB
}

func (s *levelHandler) isLastLevel() bool {
        return s.level == s.db.opt.MaxLevels-1
}

func (s *levelHandler) getTotalStaleSize() int64 {
        s.RLock()
        defer s.RUnlock()
        return s.totalStaleSize
}

func (s *levelHandler) getTotalSize() int64 {
        s.RLock()
        defer s.RUnlock()
        return s.totalSize
}

// initTables replaces s.tables with given tables. This is done during loading.
func (s *levelHandler) initTables(tables []*table.Table) {
        s.Lock()
        defer s.Unlock()

        s.tables = tables
        s.totalSize = 0
        s.totalStaleSize = 0
        for _, t := range tables {
                s.addSize(t)
        }

        if s.level == 0 {
                // Key range will overlap. Just sort by fileID in ascending order
                // because newer tables are at the end of level 0.
                sort.Slice(s.tables, func(i, j int) bool {
                        return s.tables[i].ID() < s.tables[j].ID()
                })
        } else {
                // Sort tables by keys.
                sort.Slice(s.tables, func(i, j int) bool {
                        return y.CompareKeys(s.tables[i].Smallest(), s.tables[j].Smallest()) < 0
                })
        }
}

// deleteTables remove tables idx0, ..., idx1-1.
func (s *levelHandler) deleteTables(toDel []*table.Table) error {
        s.Lock() // s.Unlock() below

        toDelMap := make(map[uint64]struct{})
        for _, t := range toDel {
                toDelMap[t.ID()] = struct{}{}
        }

        // Make a copy as iterators might be keeping a slice of tables.
        var newTables []*table.Table
        for _, t := range s.tables {
                _, found := toDelMap[t.ID()]
                if !found {
                        newTables = append(newTables, t)
                        continue
                }
                s.subtractSize(t)
        }
        s.tables = newTables

        s.Unlock() // Unlock s _before_ we DecrRef our tables, which can be slow.

        return decrRefs(toDel)
}

// replaceTables will replace tables[left:right] with newTables. Note this EXCLUDES tables[right].
// You must call decr() to delete the old tables _after_ writing the update to the manifest.
func (s *levelHandler) replaceTables(toDel, toAdd []*table.Table) error {
        // Need to re-search the range of tables in this level to be replaced as other goroutines might
        // be changing it as well.  (They can't touch our tables, but if they add/remove other tables,
        // the indices get shifted around.)
        s.Lock() // We s.Unlock() below.

        toDelMap := make(map[uint64]struct{})
        for _, t := range toDel {
                toDelMap[t.ID()] = struct{}{}
        }
        var newTables []*table.Table
        for _, t := range s.tables {
                _, found := toDelMap[t.ID()]
                if !found {
                        newTables = append(newTables, t)
                        continue
                }
                s.subtractSize(t)
        }

        // Increase totalSize first.
        for _, t := range toAdd {
                s.addSize(t)
                t.IncrRef()
                newTables = append(newTables, t)
        }

        // Assign tables.
        s.tables = newTables
        sort.Slice(s.tables, func(i, j int) bool {
                return y.CompareKeys(s.tables[i].Smallest(), s.tables[j].Smallest()) < 0
        })
        s.Unlock() // s.Unlock before we DecrRef tables -- that can be slow.
        return decrRefs(toDel)
}

// addTable adds toAdd table to levelHandler. Normally when we add tables to levelHandler, we sort
// tables based on table.Smallest. This is required for correctness of the system. But in case of
// stream writer this can be avoided. We can just add tables to levelHandler's table list
// and after all addTable calls, we can sort table list(check sortTable method).
// NOTE: levelHandler.sortTables() should be called after call addTable calls are done.
func (s *levelHandler) addTable(t *table.Table) {
        s.Lock()
        defer s.Unlock()

        s.addSize(t) // Increase totalSize first.
        t.IncrRef()
        s.tables = append(s.tables, t)
}

// sortTables sorts tables of levelHandler based on table.Smallest.
// Normally it should be called after all addTable calls.
func (s *levelHandler) sortTables() {
        s.RLock()
        defer s.RUnlock()

        sort.Slice(s.tables, func(i, j int) bool {
                return y.CompareKeys(s.tables[i].Smallest(), s.tables[j].Smallest()) < 0
        })
}

func decrRefs(tables []*table.Table) error {
        for _, table := range tables {
                if err := table.DecrRef(); err != nil {
                        return err
                }
        }
        return nil
}

func newLevelHandler(db *DB, level int) *levelHandler {
        return &levelHandler{
                level:    level,
                strLevel: fmt.Sprintf("l%d", level),
                db:       db,
        }
}

// tryAddLevel0Table returns true if ok and no stalling.
func (s *levelHandler) tryAddLevel0Table(t *table.Table) bool {
        y.AssertTrue(s.level == 0)
        // Need lock as we may be deleting the first table during a level 0 compaction.
        s.Lock()
        defer s.Unlock()
        // Stall (by returning false) if we are above the specified stall setting for L0.
        if len(s.tables) >= s.db.opt.NumLevelZeroTablesStall {
                return false
        }

        s.tables = append(s.tables, t)
        t.IncrRef()
        s.addSize(t)

        return true
}

// This should be called while holding the lock on the level.
func (s *levelHandler) addSize(t *table.Table) {
        s.totalSize += t.Size()
        s.totalStaleSize += int64(t.StaleDataSize())
}

// This should be called while holding the lock on the level.
func (s *levelHandler) subtractSize(t *table.Table) {
        s.totalSize -= t.Size()
        s.totalStaleSize -= int64(t.StaleDataSize())
}
func (s *levelHandler) numTables() int {
        s.RLock()
        defer s.RUnlock()
        return len(s.tables)
}

func (s *levelHandler) close() error {
        s.RLock()
        defer s.RUnlock()
        var err error
        for _, t := range s.tables {
                if closeErr := t.Close(-1); closeErr != nil && err == nil {
                        err = closeErr
                }
        }
        return y.Wrap(err, "levelHandler.close")
}

// getTableForKey acquires a read-lock to access s.tables. It returns a list of tableHandlers.
func (s *levelHandler) getTableForKey(key []byte) ([]*table.Table, func() error) {
        s.RLock()
        defer s.RUnlock()

        if s.level == 0 {
                // For level 0, we need to check every table. Remember to make a copy as s.tables may change
                // once we exit this function, and we don't want to lock s.tables while seeking in tables.
                // CAUTION: Reverse the tables.
                out := make([]*table.Table, 0, len(s.tables))
                for i := len(s.tables) - 1; i >= 0; i-- {
                        out = append(out, s.tables[i])
                        s.tables[i].IncrRef()
                }
                return out, func() error {
                        for _, t := range out {
                                if err := t.DecrRef(); err != nil {
                                        return err
                                }
                        }
                        return nil
                }
        }
        // For level >= 1, we can do a binary search as key range does not overlap.
        idx := sort.Search(len(s.tables), func(i int) bool {
                return y.CompareKeys(s.tables[i].Biggest(), key) >= 0
        })
        if idx >= len(s.tables) {
                // Given key is strictly > than every element we have.
                return nil, func() error { return nil }
        }
        tbl := s.tables[idx]
        tbl.IncrRef()
        return []*table.Table{tbl}, tbl.DecrRef
}

// get returns value for a given key or the key after that. If not found, return nil.
func (s *levelHandler) get(key []byte) (y.ValueStruct, error) {
        tables, decr := s.getTableForKey(key)
        keyNoTs := y.ParseKey(key)

        hash := y.Hash(keyNoTs)
        var maxVs y.ValueStruct
        for _, th := range tables {
                if th.DoesNotHave(hash) {
                        y.NumLSMBloomHitsAdd(s.db.opt.MetricsEnabled, s.strLevel, 1)
                        continue
                }

                it := th.NewIterator(0)
                defer it.Close()

                y.NumLSMGetsAdd(s.db.opt.MetricsEnabled, s.strLevel, 1)
                it.Seek(key)
                if !it.Valid() {
                        continue
                }
                if y.SameKey(key, it.Key()) {
                        if version := y.ParseTs(it.Key()); maxVs.Version < version {
                                maxVs = it.ValueCopy()
                                maxVs.Version = version
                        }
                }
        }
        return maxVs, decr()
}

// iterators returns an array of iterators, for merging.
// Note: This obtains references for the table handlers. Remember to close these iterators.
func (s *levelHandler) iterators(opt *IteratorOptions) []y.Iterator {
        s.RLock()
        defer s.RUnlock()

        var topt int
        if opt.Reverse {
                topt = table.REVERSED
        }
        if s.level == 0 {
                // Remember to add in reverse order!
                // The newer table at the end of s.tables should be added first as it takes precedence.
                // Level 0 tables are not in key sorted order, so we need to consider them one by one.
                var out []*table.Table
                for _, t := range s.tables {
                        if opt.pickTable(t) {
                                out = append(out, t)
                        }
                }
                return iteratorsReversed(out, topt)
        }

        tables := opt.pickTables(s.tables)
        if len(tables) == 0 {
                return nil
        }
        return []y.Iterator{table.NewConcatIterator(tables, topt)}
}

func (s *levelHandler) getTables(opt *IteratorOptions) []*table.Table {
        if opt.Reverse {
                panic("Invalid option for getTables")
        }

        s.RLock()
        defer s.RUnlock()

        if s.level == 0 {
                var out []*table.Table
                for _, t := range s.tables {
                        if opt.pickTable(t) {
                                t.IncrRef()
                                out = append(out, t)
                        }
                }
                return out
        }

        tables := opt.pickTables(s.tables)
        for _, t := range tables {
                t.IncrRef()
        }
        return tables
}

type levelHandlerRLocked struct{}

// overlappingTables returns the tables that intersect with key range. Returns a half-interval.
// This function should already have acquired a read lock, and this is so important the caller must
// pass an empty parameter declaring such.
func (s *levelHandler) overlappingTables(_ levelHandlerRLocked, kr keyRange) (int, int) {
        if len(kr.left) == 0 || len(kr.right) == 0 {
                return 0, 0
        }
        left := sort.Search(len(s.tables), func(i int) bool {
                return y.CompareKeys(kr.left, s.tables[i].Biggest()) <= 0
        })
        right := sort.Search(len(s.tables), func(i int) bool {
                return y.CompareKeys(kr.right, s.tables[i].Smallest()) < 0
        })
        return left, right
}

1	/*
2	* Copyright 2017 Dgraph Labs, Inc. and Contributors
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16
17	package badger
18
19	import (
20	"fmt"
21	"sort"
22	"sync"
23
24	"github.com/dgraph-io/badger/v4/table"
25	"github.com/dgraph-io/badger/v4/y"
26	)
27
28	type levelHandler struct {
29	// Guards tables, totalSize.
30	sync.RWMutex
31
32	// For level >= 1, tables are sorted by key ranges, which do not overlap.
33	// For level 0, tables are sorted by time.
34	// For level 0, newest table are at the back. Compact the oldest one first, which is at the front.
35	tables []*table.Table
36	totalSize int64
37	totalStaleSize int64
38
39	// The following are initialized once and const.
40	level int
41	strLevel string
42	db *DB
43	}
44
45	func (s *levelHandler) isLastLevel() bool {	195✔
46	return s.level == s.db.opt.MaxLevels-1	195✔
47	}	195✔
48
49	func (s *levelHandler) getTotalStaleSize() int64 {	4✔
50	s.RLock()	4✔
51	defer s.RUnlock()	4✔
52	return s.totalStaleSize	4✔
53	}	4✔
54
55	func (s *levelHandler) getTotalSize() int64 {	898,676✔
56	s.RLock()	898,676✔
57	defer s.RUnlock()	898,676✔
58	return s.totalSize	898,676✔
59	}	898,676✔
60
61	// initTables replaces s.tables with given tables. This is done during loading.
62	func (s levelHandler) initTables(tables []table.Table) {	2,170✔
63	s.Lock()	2,170✔
64	defer s.Unlock()	2,170✔
65		2,170✔
66	s.tables = tables	2,170✔
67	s.totalSize = 0	2,170✔
68	s.totalStaleSize = 0	2,170✔
69	for _, t := range tables {	2,255✔
70	s.addSize(t)	85✔
71	}	85✔
72
73	if s.level == 0 {	2,480✔
74	// Key range will overlap. Just sort by fileID in ascending order	310✔
75	// because newer tables are at the end of level 0.	310✔
76	sort.Slice(s.tables, func(i, j int) bool {	316✔
77	return s.tables[i].ID() < s.tables[j].ID()	6✔
78	})	6✔
79	} else {	1,860✔
80	// Sort tables by keys.	1,860✔
81	sort.Slice(s.tables, func(i, j int) bool {	1,951✔
82	return y.CompareKeys(s.tables[i].Smallest(), s.tables[j].Smallest()) < 0	91✔
83	})	91✔
84	}
85	}
86
87	// deleteTables remove tables idx0, ..., idx1-1.
88	func (s levelHandler) deleteTables(toDel []table.Table) error {	174✔
89	s.Lock() // s.Unlock() below	174✔
90		174✔
91	toDelMap := make(map[uint64]struct{})	174✔
92	for _, t := range toDel {	359✔
93	toDelMap[t.ID()] = struct{}{}	185✔
94	}	185✔
95
96	// Make a copy as iterators might be keeping a slice of tables.
97	var newTables []*table.Table	174✔
98	for _, t := range s.tables {	1,512✔
99	_, found := toDelMap[t.ID()]	1,338✔
100	if !found {	2,491✔
101	newTables = append(newTables, t)	1,153✔
102	continue	1,153✔
103	}
104	s.subtractSize(t)	185✔
105	}
106	s.tables = newTables	174✔
107		174✔
108	s.Unlock() // Unlock s _before_ we DecrRef our tables, which can be slow.	174✔
109		174✔
110	return decrRefs(toDel)	174✔
111	}
112
113	// replaceTables will replace tables[left:right] with newTables. Note this EXCLUDES tables[right].
114	// You must call decr() to delete the old tables _after_ writing the update to the manifest.
115	func (s levelHandler) replaceTables(toDel, toAdd []table.Table) error {	187✔
116	// Need to re-search the range of tables in this level to be replaced as other goroutines might	187✔
117	// be changing it as well. (They can't touch our tables, but if they add/remove other tables,	187✔
118	// the indices get shifted around.)	187✔
119	s.Lock() // We s.Unlock() below.	187✔
120		187✔
121	toDelMap := make(map[uint64]struct{})	187✔
122	for _, t := range toDel {	433✔
123	toDelMap[t.ID()] = struct{}{}	246✔
124	}	246✔
125	var newTables []*table.Table	187✔
126	for _, t := range s.tables {	6,917✔
127	_, found := toDelMap[t.ID()]	6,730✔
128	if !found {	13,214✔
129	newTables = append(newTables, t)	6,484✔
130	continue	6,484✔
131	}
132	s.subtractSize(t)	246✔
133	}
134
135	// Increase totalSize first.
136	for _, t := range toAdd {	648✔
137	s.addSize(t)	461✔
138	t.IncrRef()	461✔
139	newTables = append(newTables, t)	461✔
140	}	461✔
141
142	// Assign tables.
143	s.tables = newTables	187✔
144	sort.Slice(s.tables, func(i, j int) bool {	10,796✔
145	return y.CompareKeys(s.tables[i].Smallest(), s.tables[j].Smallest()) < 0	10,609✔
146	})	10,609✔
147	s.Unlock() // s.Unlock before we DecrRef tables -- that can be slow.	187✔
148	return decrRefs(toDel)	187✔
149	}
150
151	// addTable adds toAdd table to levelHandler. Normally when we add tables to levelHandler, we sort
152	// tables based on table.Smallest. This is required for correctness of the system. But in case of
153	// stream writer this can be avoided. We can just add tables to levelHandler's table list
154	// and after all addTable calls, we can sort table list(check sortTable method).
155	// NOTE: levelHandler.sortTables() should be called after call addTable calls are done.
156	func (s levelHandler) addTable(t table.Table) {	103✔
157	s.Lock()	103✔
158	defer s.Unlock()	103✔
159		103✔
160	s.addSize(t) // Increase totalSize first.	103✔
161	t.IncrRef()	103✔
162	s.tables = append(s.tables, t)	103✔
163	}	103✔
164
165	// sortTables sorts tables of levelHandler based on table.Smallest.
166	// Normally it should be called after all addTable calls.
167	func (s *levelHandler) sortTables() {	301✔
168	s.RLock()	301✔
169	defer s.RUnlock()	301✔
170		301✔
171	sort.Slice(s.tables, func(i, j int) bool {	449✔
172	return y.CompareKeys(s.tables[i].Smallest(), s.tables[j].Smallest()) < 0	148✔
173	})	148✔
174	}
175
176	func decrRefs(tables []*table.Table) error {	535✔
177	for _, table := range tables {	1,414✔
178	if err := table.DecrRef(); err != nil {	879✔
179	return err	×
180	}	×
181	}
182	return nil	535✔
183	}
184
185	func newLevelHandler(db DB, level int) levelHandler {	2,303✔
186	return &levelHandler{	2,303✔
187	level: level,	2,303✔
188	strLevel: fmt.Sprintf("l%d", level),	2,303✔
189	db: db,	2,303✔
190	}	2,303✔
191	}	2,303✔
192
193	// tryAddLevel0Table returns true if ok and no stalling.
194	func (s levelHandler) tryAddLevel0Table(t table.Table) bool {	292✔
195	y.AssertTrue(s.level == 0)	292✔
196	// Need lock as we may be deleting the first table during a level 0 compaction.	292✔
197	s.Lock()	292✔
198	defer s.Unlock()	292✔
199	// Stall (by returning false) if we are above the specified stall setting for L0.	292✔
200	if len(s.tables) >= s.db.opt.NumLevelZeroTablesStall {	292✔
201	return false	×
202	}	×
203
204	s.tables = append(s.tables, t)	292✔
205	t.IncrRef()	292✔
206	s.addSize(t)	292✔
207		292✔
208	return true	292✔
209	}
210
211	// This should be called while holding the lock on the level.
212	func (s levelHandler) addSize(t table.Table) {	941✔
213	s.totalSize += t.Size()	941✔
214	s.totalStaleSize += int64(t.StaleDataSize())	941✔
215	}	941✔
216
217	// This should be called while holding the lock on the level.
218	func (s levelHandler) subtractSize(t table.Table) {	431✔
219	s.totalSize -= t.Size()	431✔
220	s.totalStaleSize -= int64(t.StaleDataSize())	431✔
221	}	431✔
222	func (s *levelHandler) numTables() int {	126,871✔
223	s.RLock()	126,871✔
224	defer s.RUnlock()	126,871✔
225	return len(s.tables)	126,871✔
226	}	126,871✔
227
228	func (s *levelHandler) close() error {	2,240✔
229	s.RLock()	2,240✔
230	defer s.RUnlock()	2,240✔
231	var err error	2,240✔
232	for _, t := range s.tables {	2,806✔
233	if closeErr := t.Close(-1); closeErr != nil && err == nil {	566✔
234	err = closeErr	×
235	}	×
236	}
237	return y.Wrap(err, "levelHandler.close")	2,240✔
238	}
239
240	// getTableForKey acquires a read-lock to access s.tables. It returns a list of tableHandlers.
241	func (s levelHandler) getTableForKey(key []byte) ([]table.Table, func() error) {	4,934,178✔
242	s.RLock()	4,934,178✔
243	defer s.RUnlock()	4,934,178✔
244		4,934,178✔
245	if s.level == 0 {	5,639,082✔
246	// For level 0, we need to check every table. Remember to make a copy as s.tables may change	704,904✔
247	// once we exit this function, and we don't want to lock s.tables while seeking in tables.	704,904✔
248	// CAUTION: Reverse the tables.	704,904✔
249	out := make([]*table.Table, 0, len(s.tables))	704,904✔
250	for i := len(s.tables) - 1; i >= 0; i-- {	767,553✔
251	out = append(out, s.tables[i])	62,649✔
252	s.tables[i].IncrRef()	62,649✔
253	}	62,649✔
254	return out, func() error {	1,409,808✔
255	for _, t := range out {	767,553✔
256	if err := t.DecrRef(); err != nil {	62,649✔
257	return err	×
258	}	×
259	}
260	return nil	704,904✔
261	}
262	}
263	// For level >= 1, we can do a binary search as key range does not overlap.
264	idx := sort.Search(len(s.tables), func(i int) bool {	4,234,353✔
265	return y.CompareKeys(s.tables[i].Biggest(), key) >= 0	5,079✔
266	})	5,079✔
267	if idx >= len(s.tables) {	8,456,654✔
268	// Given key is strictly > than every element we have.	4,227,380✔
269	return nil, func() error { return nil }	8,454,760✔
270	}
271	tbl := s.tables[idx]	1,894✔
272	tbl.IncrRef()	1,894✔
273	return []*table.Table{tbl}, tbl.DecrRef	1,894✔
274	}
275
276	// get returns value for a given key or the key after that. If not found, return nil.
277	func (s *levelHandler) get(key []byte) (y.ValueStruct, error) {	4,934,178✔
278	tables, decr := s.getTableForKey(key)	4,934,178✔
279	keyNoTs := y.ParseKey(key)	4,934,178✔
280		4,934,178✔
281	hash := y.Hash(keyNoTs)	4,934,178✔
282	var maxVs y.ValueStruct	4,934,178✔
283	for _, th := range tables {	4,998,721✔
284	if th.DoesNotHave(hash) {	70,738✔
285	y.NumLSMBloomHitsAdd(s.db.opt.MetricsEnabled, s.strLevel, 1)	6,195✔
286	continue	6,195✔
287	}
288
289	it := th.NewIterator(0)	58,348✔
290	defer it.Close()	58,348✔
291		58,348✔
292	y.NumLSMGetsAdd(s.db.opt.MetricsEnabled, s.strLevel, 1)	58,348✔
293	it.Seek(key)	58,348✔
294	if !it.Valid() {	58,430✔
295	continue	82✔
296	}
297	if y.SameKey(key, it.Key()) {	116,390✔
298	if version := y.ParseTs(it.Key()); maxVs.Version < version {	116,241✔
299	maxVs = it.ValueCopy()	58,117✔
300	maxVs.Version = version	58,117✔
301	}	58,117✔
302	}
303	}
304	return maxVs, decr()	4,934,178✔
305	}
306
307	// iterators returns an array of iterators, for merging.
308	// Note: This obtains references for the table handlers. Remember to close these iterators.
309	func (s levelHandler) iterators(opt IteratorOptions) []y.Iterator {	286,433✔
310	s.RLock()	286,433✔
311	defer s.RUnlock()	286,433✔
312		286,433✔
313	var topt int	286,433✔
314	if opt.Reverse {	286,678✔
315	topt = table.REVERSED	245✔
316	}	245✔
317	if s.level == 0 {	327,352✔
318	// Remember to add in reverse order!	40,919✔
319	// The newer table at the end of s.tables should be added first as it takes precedence.	40,919✔
320	// Level 0 tables are not in key sorted order, so we need to consider them one by one.	40,919✔
321	var out []*table.Table	40,919✔
322	for _, t := range s.tables {	40,957✔
323	if opt.pickTable(t) {	76✔
324	out = append(out, t)	38✔
325	}	38✔
326	}
327	return iteratorsReversed(out, topt)	40,919✔
328	}
329
330	tables := opt.pickTables(s.tables)	245,514✔
331	if len(tables) == 0 {	490,937✔
332	return nil	245,423✔
333	}	245,423✔
334	return []y.Iterator{table.NewConcatIterator(tables, topt)}	91✔
335	}
336
337	func (s levelHandler) getTables(opt IteratorOptions) []*table.Table {	161✔
338	if opt.Reverse {	161✔
339	panic("Invalid option for getTables")	×
340	}
341
342	s.RLock()	161✔
343	defer s.RUnlock()	161✔
344		161✔
345	if s.level == 0 {	184✔
346	var out []*table.Table	23✔
347	for _, t := range s.tables {	25✔
348	if opt.pickTable(t) {	4✔
349	t.IncrRef()	2✔
350	out = append(out, t)	2✔
351	}	2✔
352	}
353	return out	23✔
354	}
355
356	tables := opt.pickTables(s.tables)	138✔
357	for _, t := range tables {	163✔
358	t.IncrRef()	25✔
359	}	25✔
360	return tables	138✔
361	}
362
363	type levelHandlerRLocked struct{}
364
365	// overlappingTables returns the tables that intersect with key range. Returns a half-interval.
366	// This function should already have acquired a read lock, and this is so important the caller must
367	// pass an empty parameter declaring such.
368	func (s *levelHandler) overlappingTables(_ levelHandlerRLocked, kr keyRange) (int, int) {	3,000✔
369	if len(kr.left) == 0 \|\| len(kr.right) == 0 {	3,000✔
370	return 0, 0	×
371	}	×
372	left := sort.Search(len(s.tables), func(i int) bool {	11,021✔
373	return y.CompareKeys(kr.left, s.tables[i].Biggest()) <= 0	8,021✔
374	})	8,021✔
375	right := sort.Search(len(s.tables), func(i int) bool {	11,048✔
376	return y.CompareKeys(kr.right, s.tables[i].Smallest()) < 0	8,048✔
377	})	8,048✔
378	return left, right	3,000✔
379	}

dgraph-io / badger / 5239711038

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous