forked from influxdata/influxdb
-
Notifications
You must be signed in to change notification settings - Fork 0
/
show_measurements.go
237 lines (204 loc) · 5.85 KB
/
show_measurements.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
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
package tsdb
import (
"encoding/json"
"fmt"
"sort"
"github.com/influxdb/influxdb/influxql"
)
// ShowMeasurementsExecutor implements the Executor interface for a SHOW MEASUREMENTS statement.
type ShowMeasurementsExecutor struct {
stmt *influxql.ShowMeasurementsStatement
mappers []Mapper
chunkSize int
}
// NewShowMeasurementsExecutor returns a new ShowMeasurementsExecutor.
func NewShowMeasurementsExecutor(stmt *influxql.ShowMeasurementsStatement, mappers []Mapper, chunkSize int) *ShowMeasurementsExecutor {
return &ShowMeasurementsExecutor{
stmt: stmt,
mappers: mappers,
chunkSize: chunkSize,
}
}
// Execute begins execution of the query and returns a channel to receive rows.
func (e *ShowMeasurementsExecutor) Execute() <-chan *influxql.Row {
// Create output channel and stream data in a separate goroutine.
out := make(chan *influxql.Row, 0)
go func() {
// Open the mappers.
for _, m := range e.mappers {
if err := m.Open(); err != nil {
out <- &influxql.Row{Err: err}
return
}
}
// Create a set to hold measurement names from mappers.
set := map[string]struct{}{}
// Iterate through mappers collecting measurement names.
for _, m := range e.mappers {
// Get the data from the mapper.
c, err := m.NextChunk()
if err != nil {
out <- &influxql.Row{Err: err}
return
} else if c == nil {
// Mapper had no data.
continue
}
// Convert the mapper chunk to a string array of measurement names.
mms, ok := c.([]string)
if !ok {
out <- &influxql.Row{Err: fmt.Errorf("show measurements mapper returned invalid type: %T", c)}
return
}
// Add the measurement names to the set.
for _, mm := range mms {
set[mm] = struct{}{}
}
}
// Convert the set into an array of measurement names.
measurements := make([]string, 0, len(set))
for mm := range set {
measurements = append(measurements, mm)
}
// Sort the names.
sort.Strings(measurements)
// Calculate OFFSET and LIMIT
off := e.stmt.Offset
lim := len(measurements)
stmtLim := e.stmt.Limit
if stmtLim > 0 && off+stmtLim < lim {
lim = off + stmtLim
} else if off > lim {
off, lim = 0, 0
}
// Put the results in a row and send it.
row := &influxql.Row{
Name: "measurements",
Columns: []string{"name"},
Values: make([][]interface{}, 0, len(measurements)),
}
for _, m := range measurements[off:lim] {
v := []interface{}{m}
row.Values = append(row.Values, v)
}
if len(row.Values) > 0 {
out <- row
}
close(out)
// It's important that all resources are released when execution completes.
e.close()
}()
return out
}
// Close closes the executor such that all resources are released. Once closed,
// an executor may not be re-used.
func (e *ShowMeasurementsExecutor) close() {
if e != nil {
for _, m := range e.mappers {
m.Close()
}
}
}
// ShowMeasurementsMapper is a mapper for collecting measurement names from a shard.
type ShowMeasurementsMapper struct {
remote Mapper
shard *Shard
stmt *influxql.ShowMeasurementsStatement
chunkSize int
state interface{}
}
// NewShowMeasurementsMapper returns a mapper for the given shard, which will return data for the meta statement.
func NewShowMeasurementsMapper(shard *Shard, stmt *influxql.ShowMeasurementsStatement, chunkSize int) *ShowMeasurementsMapper {
return &ShowMeasurementsMapper{
shard: shard,
stmt: stmt,
chunkSize: chunkSize,
}
}
// Open opens the mapper for use.
func (m *ShowMeasurementsMapper) Open() error {
if m.remote != nil {
return m.remote.Open()
}
var measurements Measurements
if m.shard != nil {
// If a WHERE clause was specified, filter the measurements.
if m.stmt.Condition != nil {
var err error
measurements, err = m.shard.index.measurementsByExpr(m.stmt.Condition)
if err != nil {
return err
}
} else {
// Otherwise, get all measurements from the database.
measurements = m.shard.index.Measurements()
}
sort.Sort(measurements)
}
// Create a channel to send measurement names on.
ch := make(chan string)
// Start a goroutine to send the names over the channel as needed.
go func() {
for _, mm := range measurements {
ch <- mm.Name
}
close(ch)
}()
// Store the channel as the state of the mapper.
m.state = ch
return nil
}
// SetRemote sets the remote mapper to use.
func (m *ShowMeasurementsMapper) SetRemote(remote Mapper) error {
m.remote = remote
return nil
}
// TagSets is only implemented on this mapper to satisfy the Mapper interface.
func (m *ShowMeasurementsMapper) TagSets() []string { return nil }
// Fields returns a list of field names for this mapper.
func (m *ShowMeasurementsMapper) Fields() []string { return []string{"name"} }
// NextChunk returns the next chunk of measurement names.
func (m *ShowMeasurementsMapper) NextChunk() (interface{}, error) {
if m.remote != nil {
b, err := m.remote.NextChunk()
if err != nil {
return nil, err
} else if b == nil {
return nil, nil
}
names := []string{}
if err := json.Unmarshal(b.([]byte), &names); err != nil {
return nil, err
} else if len(names) == 0 {
// Mapper on other node sent 0 values so it's done.
return nil, nil
}
return names, nil
}
return m.nextChunk()
}
// nextChunk implements next chunk logic for a local shard.
func (m *ShowMeasurementsMapper) nextChunk() (interface{}, error) {
// Allocate array to hold measurement names.
names := make([]string, 0, m.chunkSize)
// Get the channel of measurement names from the state.
measurementNames := m.state.(chan string)
// Get the next chunk of names.
for n := range measurementNames {
names = append(names, n)
if len(names) == m.chunkSize {
break
}
}
// See if we've read all the names.
if len(names) == 0 {
return nil, nil
}
return names, nil
}
// Close closes the mapper.
func (m *ShowMeasurementsMapper) Close() {
if m.remote != nil {
m.remote.Close()
}
}