Skip to content

Commit 8a588af

Browse files
Likith101Likith101
committed
MB-68045: Toy: Added a Merge Enumerator
1 parent 4e38ae4 commit 8a588af

4 files changed

Lines changed: 313 additions & 119 deletions

File tree

‎enumerator.go‎

Lines changed: 12 additions & 116 deletions
Original file line numberDiff line numberDiff line change
@@ -14,125 +14,21 @@
1414

1515
package zap
1616

17-
import (
18-
"bytes"
17+
import "github.com/blevesearch/vellum"
1918

20-
"github.com/blevesearch/vellum"
21-
)
19+
var maxItrsPerBasicEnum = 64
2220

23-
// enumerator provides an ordered traversal of multiple vellum
24-
// iterators. Like JOIN of iterators, the enumerator produces a
25-
// sequence of (key, iteratorIndex, value) tuples, sorted by key ASC,
26-
// then iteratorIndex ASC, where the same key might be seen or
27-
// repeated across multiple child iterators.
28-
type enumerator struct {
29-
itrs []vellum.Iterator
30-
currKs [][]byte
31-
currVs []uint64
32-
33-
lowK []byte
34-
lowIdxs []int
35-
lowCurr int
36-
}
37-
38-
// newEnumerator returns a new enumerator over the vellum Iterators
39-
func newEnumerator(itrs []vellum.Iterator) (*enumerator, error) {
40-
rv := &enumerator{
41-
itrs: itrs,
42-
currKs: make([][]byte, len(itrs)),
43-
currVs: make([]uint64, len(itrs)),
44-
lowIdxs: make([]int, 0, len(itrs)),
45-
}
46-
for i, itr := range rv.itrs {
47-
rv.currKs[i], rv.currVs[i] = itr.Current()
48-
}
49-
rv.updateMatches(false)
50-
if rv.lowK == nil && len(rv.lowIdxs) == 0 {
51-
return rv, vellum.ErrIteratorDone
52-
}
53-
return rv, nil
54-
}
55-
56-
// updateMatches maintains the low key matches based on the currKs
57-
func (m *enumerator) updateMatches(skipEmptyKey bool) {
58-
m.lowK = nil
59-
m.lowIdxs = m.lowIdxs[:0]
60-
m.lowCurr = 0
61-
62-
for i, key := range m.currKs {
63-
if (key == nil && m.currVs[i] == 0) || // in case of empty iterator
64-
(len(key) == 0 && skipEmptyKey) { // skip empty keys
65-
continue
66-
}
67-
68-
cmp := bytes.Compare(key, m.lowK)
69-
if cmp < 0 || len(m.lowIdxs) == 0 {
70-
// reached a new low
71-
m.lowK = key
72-
m.lowIdxs = m.lowIdxs[:0]
73-
m.lowIdxs = append(m.lowIdxs, i)
74-
} else if cmp == 0 {
75-
m.lowIdxs = append(m.lowIdxs, i)
76-
}
77-
}
78-
}
79-
80-
// Current returns the enumerator's current key, iterator-index, and
81-
// value. If the enumerator is not pointing at a valid value (because
82-
// Next returned an error previously), Current will return nil,0,0.
83-
func (m *enumerator) Current() ([]byte, int, uint64) {
84-
var i int
85-
var v uint64
86-
if m.lowCurr < len(m.lowIdxs) {
87-
i = m.lowIdxs[m.lowCurr]
88-
v = m.currVs[i]
89-
}
90-
return m.lowK, i, v
91-
}
92-
93-
// GetLowIdxsAndValues will return all of the iterator indices
94-
// which point to the current key, and their corresponding
95-
// values. This can be used by advanced caller which may need
96-
// to peek into these other sets of data before processing.
97-
func (m *enumerator) GetLowIdxsAndValues() ([]int, []uint64) {
98-
values := make([]uint64, 0, len(m.lowIdxs))
99-
for _, idx := range m.lowIdxs {
100-
values = append(values, m.currVs[idx])
101-
}
102-
return m.lowIdxs, values
103-
}
104-
105-
// Next advances the enumerator to the next key/iterator/value result,
106-
// else vellum.ErrIteratorDone is returned.
107-
func (m *enumerator) Next() error {
108-
m.lowCurr += 1
109-
if m.lowCurr >= len(m.lowIdxs) {
110-
// move all the current low iterators forwards
111-
for _, vi := range m.lowIdxs {
112-
err := m.itrs[vi].Next()
113-
if err != nil && err != vellum.ErrIteratorDone {
114-
return err
115-
}
116-
m.currKs[vi], m.currVs[vi] = m.itrs[vi].Current()
117-
}
118-
// can skip any empty keys encountered at this point
119-
m.updateMatches(true)
120-
}
121-
if m.lowK == nil && len(m.lowIdxs) == 0 {
122-
return vellum.ErrIteratorDone
123-
}
124-
return nil
21+
type enumerator interface {
22+
Current() ([]byte, int, uint64)
23+
GetLowIdxsAndValues() ([]int, []uint64)
24+
Next() error
25+
Close() error
12526
}
12627

127-
// Close all the underlying Iterators. The first error, if any, will
128-
// be returned.
129-
func (m *enumerator) Close() error {
130-
var rv error
131-
for _, itr := range m.itrs {
132-
err := itr.Close()
133-
if rv == nil {
134-
rv = err
135-
}
28+
func newEnumerator(itrs []vellum.Iterator) (enumerator, error) {
29+
if len(itrs) < maxItrsPerBasicEnum {
30+
return newBasicEnumerator(itrs, 0)
31+
} else {
32+
return newMergeEnumerator(itrs)
13633
}
137-
return rv
13834
}

‎enumerator_basic.go‎

Lines changed: 140 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,140 @@
1+
// Copyright (c) 2024 Couchbase, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package zap
16+
17+
import (
18+
"bytes"
19+
20+
"github.com/blevesearch/vellum"
21+
)
22+
23+
// enumerator provides an ordered traversal of multiple vellum
24+
// iterators. Like JOIN of iterators, the enumerator produces a
25+
// sequence of (key, iteratorIndex, value) tuples, sorted by key ASC,
26+
// then iteratorIndex ASC, where the same key might be seen or
27+
// repeated across multiple child iterators.
28+
type basicEnumerator struct {
29+
itrs []vellum.Iterator
30+
currKs [][]byte
31+
currVs []uint64
32+
id int
33+
34+
lowK []byte
35+
lowIdxs []int
36+
lowCurr int
37+
}
38+
39+
// newEnumerator returns a new enumerator over the vellum Iterators
40+
func newBasicEnumerator(itrs []vellum.Iterator, id int) (enumerator, error) {
41+
rv := &basicEnumerator{
42+
itrs: itrs,
43+
currKs: make([][]byte, len(itrs)),
44+
currVs: make([]uint64, len(itrs)),
45+
lowIdxs: make([]int, 0, len(itrs)),
46+
id: id,
47+
}
48+
for i, itr := range rv.itrs {
49+
rv.currKs[i], rv.currVs[i] = itr.Current()
50+
}
51+
rv.updateMatches(false)
52+
if rv.lowK == nil && len(rv.lowIdxs) == 0 {
53+
return rv, vellum.ErrIteratorDone
54+
}
55+
return rv, nil
56+
}
57+
58+
// updateMatches maintains the low key matches based on the currKs
59+
func (m *basicEnumerator) updateMatches(skipEmptyKey bool) {
60+
m.lowK = nil
61+
m.lowIdxs = m.lowIdxs[:0]
62+
m.lowCurr = 0
63+
64+
for i, key := range m.currKs {
65+
if (key == nil && m.currVs[i] == 0) || // in case of empty iterator
66+
(len(key) == 0 && skipEmptyKey) { // skip empty keys
67+
continue
68+
}
69+
70+
cmp := bytes.Compare(key, m.lowK)
71+
if cmp < 0 || len(m.lowIdxs) == 0 {
72+
// reached a new low
73+
m.lowK = key
74+
m.lowIdxs = m.lowIdxs[:0]
75+
m.lowIdxs = append(m.lowIdxs, i+m.id*maxItrsPerBasicEnum)
76+
} else if cmp == 0 {
77+
m.lowIdxs = append(m.lowIdxs, i+m.id*maxItrsPerBasicEnum)
78+
}
79+
}
80+
}
81+
82+
// Current returns the enumerator's current key, iterator-index, and
83+
// value. If the enumerator is not pointing at a valid value (because
84+
// Next returned an error previously), Current will return nil,0,0.
85+
func (m *basicEnumerator) Current() ([]byte, int, uint64) {
86+
var i int
87+
var v uint64
88+
if m.lowCurr < len(m.lowIdxs) {
89+
i = m.lowIdxs[m.lowCurr]
90+
v = m.currVs[i%maxItrsPerBasicEnum]
91+
}
92+
return m.lowK, i, v
93+
}
94+
95+
// GetLowIdxsAndValues will return all of the iterator indices
96+
// which point to the current key, and their corresponding
97+
// values. This can be used by advanced caller which may need
98+
// to peek into these other sets of data before processing.
99+
func (m *basicEnumerator) GetLowIdxsAndValues() ([]int, []uint64) {
100+
values := make([]uint64, 0, len(m.lowIdxs))
101+
for _, idx := range m.lowIdxs {
102+
values = append(values, m.currVs[idx%maxItrsPerBasicEnum])
103+
}
104+
return m.lowIdxs, values
105+
}
106+
107+
// Next advances the enumerator to the next key/iterator/value result,
108+
// else vellum.ErrIteratorDone is returned.
109+
func (m *basicEnumerator) Next() error {
110+
m.lowCurr += 1
111+
if m.lowCurr >= len(m.lowIdxs) {
112+
// move all the current low iterators forwards
113+
for _, vi := range m.lowIdxs {
114+
err := m.itrs[vi%maxItrsPerBasicEnum].Next()
115+
if err != nil && err != vellum.ErrIteratorDone {
116+
return err
117+
}
118+
m.currKs[vi%maxItrsPerBasicEnum], m.currVs[vi%maxItrsPerBasicEnum] = m.itrs[vi%maxItrsPerBasicEnum].Current()
119+
}
120+
// can skip any empty keys encountered at this point
121+
m.updateMatches(true)
122+
}
123+
if m.lowK == nil && len(m.lowIdxs) == 0 {
124+
return vellum.ErrIteratorDone
125+
}
126+
return nil
127+
}
128+
129+
// Close all the underlying Iterators. The first error, if any, will
130+
// be returned.
131+
func (m *basicEnumerator) Close() error {
132+
var rv error
133+
for _, itr := range m.itrs {
134+
err := itr.Close()
135+
if rv == nil {
136+
rv = err
137+
}
138+
}
139+
return rv
140+
}

0 commit comments

Comments
 (0)