-
Notifications
You must be signed in to change notification settings - Fork 342
/
merge_iterator.go
207 lines (168 loc) · 4.37 KB
/
merge_iterator.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
package onboard
import (
"errors"
"strings"
"github.com/treeverse/lakefs/pkg/catalog"
)
// prefixMergeIterator takes inventory iterator and commit iterator and merges them.
// If an entry's prefix is in `prefixes`, it would be taken from invIt.
// No entry from committedIt would be returned if its prefix is in prefixes.
// The prefixes behaviour guarantees that the two iterators are distinct.
type prefixMergeIterator struct {
invIt catalog.EntryIterator
committedIt catalog.EntryIterator
err error
bothDone bool
initialized bool
}
var ErrNotSeekable = errors.New("iterator isn't seekable")
func newPrefixMergeIterator(invIt catalog.EntryIterator, committedIt catalog.EntryListingIterator, prefixes []string) catalog.EntryIterator {
if len(prefixes) == 0 {
// If there isn't a filter, take everything from the inventory
return invIt
}
return &prefixMergeIterator{
invIt: newPrefixIterator(invIt, prefixes, true),
committedIt: newPrefixIterator(&listToEntryIterator{EntryListingIterator: committedIt}, prefixes, false),
bothDone: false,
initialized: false,
}
}
func (pmi *prefixMergeIterator) Next() bool {
if pmi.err != nil {
return false
}
if !pmi.initialized {
pmi.initialized = true
invHasMore := pmi.advanceIt(pmi.invIt)
committedHasMore := pmi.advanceIt(pmi.committedIt)
return invHasMore || committedHasMore
}
switch {
case pmi.invIt.Value() == nil && pmi.committedIt.Value() == nil:
pmi.bothDone = true
return false
case pmi.invIt.Value() == nil:
return pmi.advanceIt(pmi.committedIt)
case pmi.committedIt.Value() == nil:
return pmi.advanceIt(pmi.invIt)
// from now on - both iterators have values,
// so we'll always return true
case strings.Compare(pmi.committedIt.Value().Path.String(), pmi.invIt.Value().Path.String()) < 0:
_ = pmi.advanceIt(pmi.committedIt)
return true
default:
_ = pmi.advanceIt(pmi.invIt)
return true
}
}
func (pmi *prefixMergeIterator) advanceIt(it catalog.EntryIterator) bool {
hasNext := it.Next()
pmi.err = it.Err()
return hasNext
}
func (pmi *prefixMergeIterator) SeekGE(_ catalog.Path) {
pmi.err = ErrNotSeekable
}
func (pmi *prefixMergeIterator) Value() *catalog.EntryRecord {
if pmi.err != nil || pmi.bothDone {
return nil
}
committedVal := pmi.committedIt.Value()
invVal := pmi.invIt.Value()
if committedVal == nil {
return invVal
}
if invVal == nil {
return committedVal
}
if strings.Compare(committedVal.Path.String(), invVal.Path.String()) < 0 {
return committedVal
}
return invVal
}
func (pmi *prefixMergeIterator) Err() error {
return pmi.err
}
func (pmi *prefixMergeIterator) Close() {
pmi.invIt.Close()
if pmi.err != nil {
pmi.err = pmi.invIt.Err()
}
pmi.committedIt.Close()
if pmi.err != nil {
pmi.err = pmi.committedIt.Err()
}
}
type prefixIterator struct {
it catalog.EntryIterator
prefixes []string
err error
value *catalog.EntryRecord
allowPrefixes bool
}
func newPrefixIterator(it catalog.EntryIterator, prefixes []string, allowPrefixes bool) catalog.EntryIterator {
return &prefixIterator{
it: it,
prefixes: prefixes,
allowPrefixes: allowPrefixes,
}
}
func (pi *prefixIterator) Next() bool {
if pi.err != nil {
return false
}
for pi.it.Next() {
// iterate until finding the matching an entry
// that doesn't start with one of the prefixes.
val := pi.it.Value()
if pi.include(val) {
pi.value = &catalog.EntryRecord{
Path: val.Path,
Entry: val.Entry,
}
return true
}
}
// reached the end of the iterator
pi.err = pi.it.Err()
pi.value = nil
return false
}
func (pi *prefixIterator) include(val *catalog.EntryRecord) bool {
for _, p := range pi.prefixes {
if strings.HasPrefix(val.Path.String(), p) {
return pi.allowPrefixes
}
}
return !pi.allowPrefixes
}
func (pi *prefixIterator) SeekGE(_ catalog.Path) {
pi.err = ErrNotSeekable
}
func (pi *prefixIterator) Value() *catalog.EntryRecord {
if pi.err != nil {
return nil
}
return pi.value
}
func (pi *prefixIterator) Err() error {
return pi.err
}
func (pi *prefixIterator) Close() {
pi.it.Close()
pi.err = pi.it.Err()
}
type listToEntryIterator struct {
catalog.EntryListingIterator
}
func (iter *listToEntryIterator) Value() *catalog.EntryRecord {
val := iter.EntryListingIterator.Value()
if val == nil {
return nil
}
return &catalog.EntryRecord{
Path: val.Path,
Entry: val.Entry,
}
}