/
query_impl.go
158 lines (147 loc) · 3.06 KB
/
query_impl.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
package query
import (
"path"
goprocess "github.com/jbenet/goprocess"
)
// NaiveFilter applies a filter to the results.
func NaiveFilter(qr Results, filter Filter) Results {
return ResultsFromIterator(qr.Query(), Iterator{
Next: func() (Result, bool) {
for {
e, ok := qr.NextSync()
if !ok {
return Result{}, false
}
if e.Error != nil || filter.Filter(e.Entry) {
return e, true
}
}
},
Close: func() error {
return qr.Close()
},
})
}
// NaiveLimit truncates the results to a given int limit
func NaiveLimit(qr Results, limit int) Results {
if limit == 0 {
// 0 means no limit
return qr
}
closed := false
return ResultsFromIterator(qr.Query(), Iterator{
Next: func() (Result, bool) {
if limit == 0 {
if !closed {
closed = true
err := qr.Close()
if err != nil {
return Result{Error: err}, true
}
}
return Result{}, false
}
limit--
return qr.NextSync()
},
Close: func() error {
if closed {
return nil
}
closed = true
return qr.Close()
},
})
}
// NaiveOffset skips a given number of results
func NaiveOffset(qr Results, offset int) Results {
return ResultsFromIterator(qr.Query(), Iterator{
Next: func() (Result, bool) {
for ; offset > 0; offset-- {
res, ok := qr.NextSync()
if !ok || res.Error != nil {
return res, ok
}
}
return qr.NextSync()
},
Close: func() error {
return qr.Close()
},
})
}
// NaiveOrder reorders results according to given orders.
// WARNING: this is the only non-stream friendly operation!
func NaiveOrder(qr Results, orders ...Order) Results {
// Short circuit.
if len(orders) == 0 {
return qr
}
return ResultsWithProcess(qr.Query(), func(worker goprocess.Process, out chan<- Result) {
defer qr.Close()
var entries []Entry
collect:
for {
select {
case <-worker.Closing():
return
case e, ok := <-qr.Next():
if !ok {
break collect
}
if e.Error != nil {
out <- e
continue
}
entries = append(entries, e.Entry)
}
}
Sort(orders, entries)
for _, e := range entries {
select {
case <-worker.Closing():
return
case out <- Result{Entry: e}:
}
}
})
}
func NaiveQueryApply(q Query, qr Results) Results {
if q.Prefix != "" {
// Clean the prefix as a key and append / so a prefix of /bar
// only finds /bar/baz, not /barbaz.
prefix := q.Prefix
if len(prefix) == 0 {
prefix = "/"
} else {
if prefix[0] != '/' {
prefix = "/" + prefix
}
prefix = path.Clean(prefix)
}
// If the prefix is empty, ignore it.
if prefix != "/" {
qr = NaiveFilter(qr, FilterKeyPrefix{prefix + "/"})
}
}
for _, f := range q.Filters {
qr = NaiveFilter(qr, f)
}
if len(q.Orders) > 0 {
qr = NaiveOrder(qr, q.Orders...)
}
if q.Offset != 0 {
qr = NaiveOffset(qr, q.Offset)
}
if q.Limit != 0 {
qr = NaiveLimit(qr, q.Limit)
}
return qr
}
func ResultEntriesFrom(keys []string, vals [][]byte) []Entry {
re := make([]Entry, len(keys))
for i, k := range keys {
re[i] = Entry{Key: k, Size: len(vals[i]), Value: vals[i]}
}
return re
}