-
Notifications
You must be signed in to change notification settings - Fork 342
/
commit_ordered_iterator.go
142 lines (126 loc) · 3.65 KB
/
commit_ordered_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
package ref
import (
"context"
"errors"
sq "github.com/Masterminds/squirrel"
"github.com/treeverse/lakefs/pkg/db"
"github.com/treeverse/lakefs/pkg/graveler"
)
type OrderedCommitIteratorOption func(oci *OrderedCommitIterator)
// WithOnlyAncestryLeaves causes the iterator to return only commits which are not the first parent of any other commit.
// Consider a commit graph where all non-first-parent edges are removed. This graph is a tree, and ancestry leaves are its leaves.
func WithOnlyAncestryLeaves() OrderedCommitIteratorOption {
return func(oci *OrderedCommitIterator) {
oci.onlyAncestryLeaves = true
}
}
// NewOrderedCommitIterator returns an iterator over all commits in the given repository.
// Ordering is based on the Commit ID value.
func NewOrderedCommitIterator(ctx context.Context, database db.Database, repositoryID graveler.RepositoryID, prefetchSize int, opts ...OrderedCommitIteratorOption) *OrderedCommitIterator {
res := &OrderedCommitIterator{
ctx: ctx,
db: database,
repositoryID: repositoryID,
prefetchSize: prefetchSize,
buf: make([]*graveler.CommitRecord, 0, prefetchSize),
}
for _, opt := range opts {
opt(res)
}
return res
}
type OrderedCommitIterator struct {
ctx context.Context
db db.Database
repositoryID graveler.RepositoryID
prefetchSize int
buf []*graveler.CommitRecord
err error
value *graveler.CommitRecord
offset string
state iteratorState
onlyAncestryLeaves bool
}
func (iter *OrderedCommitIterator) Next() bool {
if iter.err != nil {
return false
}
iter.maybeFetch()
// stage a value and increment offset
if len(iter.buf) == 0 {
return false
}
iter.value = iter.buf[0]
iter.buf = iter.buf[1:]
iter.offset = string(iter.value.CommitID)
return true
}
func (iter *OrderedCommitIterator) maybeFetch() {
if iter.state == iteratorStateDone {
return
}
if len(iter.buf) > 0 {
return
}
psql := sq.StatementBuilder.PlaceholderFormat(sq.Dollar)
q := psql.Select("id", "committer", "message", "creation_date", "meta_range_id", "parents", "metadata", "version").
From("graveler_commits").
Where(sq.Eq{"repository_id": iter.repositoryID})
if iter.state == iteratorStateInit {
iter.state = iteratorStateQuerying
q = q.Where(sq.GtOrEq{"id": iter.offset})
} else {
q = q.Where(sq.Gt{"id": iter.offset})
}
var buf []*commitRecord
if iter.onlyAncestryLeaves {
notExistsCondition := psql.Select("*").
Prefix("NOT EXISTS (").Suffix(")").
From("graveler_commits c2").
Where("c2.repository_id=graveler_commits.repository_id").
Where("first_parent(c2.parents, c2.version)=graveler_commits.id")
q = q.Where(notExistsCondition)
}
q = q.OrderBy("id ASC")
q = q.Limit(uint64(iter.prefetchSize))
query, args, err := q.ToSql()
if err != nil {
iter.err = err
return
}
err = iter.db.Select(iter.ctx, &buf, query, args...)
if err != nil {
iter.err = err
return
}
if len(buf) < iter.prefetchSize {
iter.state = iteratorStateDone
}
for _, c := range buf {
rec := c.toGravelerCommitRecord()
iter.buf = append(iter.buf, rec)
}
}
func (iter *OrderedCommitIterator) SeekGE(id graveler.CommitID) {
if errors.Is(iter.err, ErrIteratorClosed) {
return
}
iter.offset = string(id)
iter.state = iteratorStateInit
iter.buf = iter.buf[:0]
iter.value = nil
iter.err = nil
}
func (iter *OrderedCommitIterator) Value() *graveler.CommitRecord {
if iter.err != nil {
return nil
}
return iter.value
}
func (iter *OrderedCommitIterator) Err() error {
return iter.err
}
func (iter *OrderedCommitIterator) Close() {
iter.err = ErrIteratorClosed
iter.buf = nil
}