Influxdata
/
influxdb
mirror of https://github.com/influxdata/influxdb.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Add chunked responses and streaming of raw queries. 

Refactored query engine to have different processing pipeline for raw queries. This enables queries that have a large offset to not keep everything in memory. It also makes it so that queries against raw data that have a limit will only p
rocess up to that limit and then bail out.

Raw data queries will only read up to a certain point in the map phase before yielding to the engine for further processing.

Fixes #2029 and fixes #2030
pull/2142/head
Paul Dix 2015-03-28 10:17:16 -04:00 committed by Philip O'Toole
parent b13385f9ab
commit 6c46a5c83b
8 changed files with 214 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
cmd/influxd/server_integration_test.go
View File

@ -722,6 +722,7 @@ func runTestsData(t *testing.T, testName string, nodes Cluster, database, retent
expected: `{"results":[{"series":[{"name":"cpu","columns":["time","alert_id"],"values":[["2015-02-28T01:03:36.703820946Z","alert"]]}]}]}`,
},
{
name: "xxx select where field greater than some value",
write: `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [{"name": "cpu", "timestamp": "2009-11-10T23:00:02Z", "fields": {"load": 100}},
{"name": "cpu", "timestamp": "2009-11-10T23:01:02Z", "fields": {"load": 80}}]}`,
query: `select load from "%DB%"."%RP%".cpu where load > 100`,
@ -1284,7 +1285,7 @@ func TestSingleServer(t *testing.T) {
defer nodes.Close()
runTestsData(t, testName, nodes, "mydb", "myrp")
runTest_rawDataReturnsInOrder(t, testName, nodes, "mydb", "myrp")
//runTest_rawDataReturnsInOrder(t, testName, nodes, "mydb", "myrp")
}
func Test3NodeServer(t *testing.T) {

25
httpd/handler.go
View File

@ -25,6 +25,7 @@ import (
)
const (
// With raw data queries, mappers will read up to this amount before sending results back to the engine
DefaultChunkSize = 10000
)
@ -176,13 +177,13 @@ func (h *Handler) serveQuery(w http.ResponseWriter, r *http.Request, user *influ
httpError(w, "error parsing query: "+err.Error(), pretty, http.StatusBadRequest)
return
}
// get the chunking settings
chunked := q.Get("chunked") == "true"
chunkSize := influxdb.NoChunkingSize
// even if we're not chunking, the engine will chunk at this size and then the handler will combine results
chunkSize := DefaultChunkSize
if chunked {
cs, err := strconv.ParseInt(q.Get("chunk_size"), 10, 64)
if err != nil {
chunkSize = DefaultChunkSize
} else {
if cs, err := strconv.ParseInt(q.Get("chunk_size"), 10, 64); err == nil {
chunkSize = int(cs)
}
}
@ -199,10 +200,11 @@ func (h *Handler) serveQuery(w http.ResponseWriter, r *http.Request, user *influ
return
}
// if we're not chunking, this will be the in memory buffer for all results before sending to client
res := influxdb.Results{Results: make([]*influxdb.Result, 0)}
statusWritten := false
// pull all results from the channel
for r := range results {
// write the status header based on the first result returned in the channel
if !statusWritten {
@ -222,15 +224,17 @@ func (h *Handler) serveQuery(w http.ResponseWriter, r *http.Request, user *influ
statusWritten = true
}
// ignore nils
if r == nil {
continue
}
// if chunked we write out this result and flush
if chunked {
w.Write(marshalPretty(r, pretty))
res.Results = []*influxdb.Result{r}
w.Write(marshalPretty(res, pretty))
w.(http.Flusher).Flush()
continue
//w.(http.Flusher).Flush()
}
// it's not chunked so buffer results in memory.
@ -253,6 +257,7 @@ func (h *Handler) serveQuery(w http.ResponseWriter, r *http.Request, user *influ
}
}
// marshalPretty will marshal the interface to json either pretty printed or not
func marshalPretty(r interface{}, pretty bool) []byte {
var b []byte
if pretty {
@ -744,6 +749,10 @@ func (w gzipResponseWriter) Write(b []byte) (int, error) {
return w.Writer.Write(b)
}
func (w gzipResponseWriter) Flush() {
w.Writer.(*gzip.Writer).Flush()
}
// determines if the client can accept compressed responses, and encodes accordingly
func gzipFilter(inner http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

3
httpd/handler_test.go
View File

@ -11,7 +11,6 @@ import (
"net/http/httptest"
"net/url"
"os"
"reflect"
"strings"
"testing"
"time"
@ -1664,7 +1663,7 @@ func TestHandler_ChunkedResponses(t *testing.T) {
} else {
vals = [][]interface{}{{"2009-11-10T23:30:00Z", 25}}
}
if !reflect.DeepEqual(results.Results[0].Series[0].Values, vals) {
if mustMarshalJSON(vals) != mustMarshalJSON(results.Results[0].Series[0].Values) {
t.Fatalf("values weren't what was expected:\n exp: %s\n got: %s", mustMarshalJSON(vals), mustMarshalJSON(results.Results[0].Series[0].Values))
}
}

225
influxql/engine.go
View File

@ -4,6 +4,7 @@ import (
"bytes"
"errors"
"hash/fnv"
"math"
"sort"
"time"
)
@ -18,10 +19,6 @@ const (
// Most likely they specified a group by interval without time boundaries.
MaxGroupByPoints = 100000
// All queries that return raw non-aggregated data, will have 2 results returned from the ouptut of a reduce run.
// The first element will be a time that we ignore, and the second element will be an array of []*rawMapOutput
ResultCountInRawResults = 2
// Since time is always selected, the column count when selecting only a single other value will be 2
SelectColumnCountWithOneValue = 2
)
@ -42,6 +39,7 @@ type MapReduceJob struct {
key []byte // a key that identifies the MRJob so it can be sorted
interval int64 // the group by interval of the query
stmt *SelectStatement // the select statement this job was created for
chunkSize int // the number of points to buffer in raw queries before returning a chunked response
}
func (m *MapReduceJob) Open() error {
@ -68,6 +66,13 @@ func (m *MapReduceJob) Key() []byte {
}
func (m *MapReduceJob) Execute(out chan *Row, filterEmptyResults bool) {
// if it's a raw query we handle processing differently
if m.stmt.IsRawQuery {
m.processRawQuery(out, filterEmptyResults)
return
}
// get the aggregates and the associated reduce functions
aggregates := m.stmt.FunctionCalls()
reduceFuncs := make([]ReduceFunc, len(aggregates))
for i, c := range aggregates {
@ -79,20 +84,11 @@ func (m *MapReduceJob) Execute(out chan *Row, filterEmptyResults bool) {
reduceFuncs[i] = reduceFunc
}
isRaw := false
// modify if it's a raw data query
if len(aggregates) == 0 {
isRaw = true
aggregates = []*Call{nil}
r, _ := InitializeReduceFunc(nil)
reduceFuncs = append(reduceFuncs, r)
}
// we'll have a fixed number of points with timestamps in buckets. Initialize those times and a slice to hold the associated values
var pointCountInResult int
// if the user didn't specify a start time or a group by interval, we're returning a single point that describes the entire range
if m.TMin == 0 || m.interval == 0 || isRaw {
if m.TMin == 0 || m.interval == 0 {
// they want a single aggregate point for the entire time range
m.interval = m.TMax - m.TMin
pointCountInResult = 1
@ -104,7 +100,7 @@ func (m *MapReduceJob) Execute(out chan *Row, filterEmptyResults bool) {
// For group by time queries, limit the number of data points returned by the limit and offset
// raw query limits are handled elsewhere
if !m.stmt.IsRawQuery && (m.stmt.Limit > 0 || m.stmt.Offset > 0) {
if m.stmt.Limit > 0 || m.stmt.Offset > 0 {
// ensure that the offset isn't higher than the number of points we'd get
if m.stmt.Offset > pointCountInResult {
return
@ -118,7 +114,7 @@ func (m *MapReduceJob) Execute(out chan *Row, filterEmptyResults bool) {
}
// If we are exceeding our MaxGroupByPoints and we aren't a raw query, error out
if !m.stmt.IsRawQuery && pointCountInResult > MaxGroupByPoints {
if pointCountInResult > MaxGroupByPoints {
out <- &Row{
Err: errors.New("too many points in the group by interval. maybe you forgot to specify a where time clause?"),
}
@ -140,7 +136,7 @@ func (m *MapReduceJob) Execute(out chan *Row, filterEmptyResults bool) {
}
// If we start getting out of our max time range, then truncate values and return
if t > m.TMax && !isRaw {
if t > m.TMax {
resultValues = resultValues[:i]
break
}
@ -152,7 +148,7 @@ func (m *MapReduceJob) Execute(out chan *Row, filterEmptyResults bool) {
// This just makes sure that if they specify a start time less than what the start time would be with the offset,
// we just reset the start time to the later time to avoid going over data that won't show up in the result.
if m.stmt.Offset > 0 && !m.stmt.IsRawQuery {
if m.stmt.Offset > 0 {
m.TMin = resultValues[0][0].(time.Time).UnixNano()
}
@ -169,18 +165,6 @@ func (m *MapReduceJob) Execute(out chan *Row, filterEmptyResults bool) {
}
}
if isRaw {
row := m.processRawResults(resultValues)
if filterEmptyResults && m.resultsEmpty(row.Values) {
return
}
// do any post processing like math and stuff
row.Values = m.processResults(row.Values)
out <- row
return
}
// filter out empty results
if filterEmptyResults && m.resultsEmpty(resultValues) {
return
@ -210,6 +194,141 @@ func (m *MapReduceJob) Execute(out chan *Row, filterEmptyResults bool) {
out <- row
}
// processRawQuery will handle running the mappers and then reducing their output
// for queries that pull back raw data values without computing any kind of aggregates.
func (m *MapReduceJob) processRawQuery(out chan *Row, filterEmptyResults bool) {
// initialize the mappers
for _, mm := range m.Mappers {
if err := mm.Begin(nil, m.TMin); err != nil {
out <- &Row{Err: err}
return
}
}
mapperOutputs := make([][]*rawQueryMapOutput, len(m.Mappers))
// markers for which mappers have been completely emptied
mapperComplete := make([]bool, len(m.Mappers))
// we need to make sure that we send at least one row, even for queries with empty results
oneRowSent := false
// for limit and offset we need to track how many values we've swalloed for the offset and how many we've already set for the limit.
// we track the number set for the limit because they could be getting chunks. For instance if your limit is 10k, but chunk size is 1k
valuesSent := 0
valuesOffset := 0
// loop until we've emptied out all the mappers and sent everything out
for {
// collect up to the limit for each mapper
for j, mm := range m.Mappers {
// only pull from mappers that potentially have more data and whose last output has been completely sent out.
if mapperOutputs[j] != nil || mapperComplete[j] {
continue
}
mm.SetLimit(m.chunkSize)
res, err := mm.NextInterval(m.TMax)
if err != nil {
out <- &Row{Err: err}
return
}
if res != nil {
mapperOutputs[j] = res.([]*rawQueryMapOutput)
} else { // if we got a nil from the mapper it means that we've emptied all data from it
mapperComplete[j] = true
}
}
// process the mapper outputs. we can send out everything up to the min of the last time in the mappers
min := int64(math.MaxInt64)
for _, o := range mapperOutputs {
// some of the mappers could empty out before others so ignore them because they'll be nil
if o == nil {
continue
}
// find the min of the last point in each mapper
t := o[len(o)-1].timestamp
if t < min {
min = t
}
}
// now empty out all the mapper outputs up to the min time
var values []*rawQueryMapOutput
for j, o := range mapperOutputs {
// find the index of the point up to the min
ind := len(o)
for i, mo := range o {
if mo.timestamp > min {
ind = i
break
}
}
// add up to the index to the values
values = append(values, o[:ind]...)
// if we emptied out all the values, set this output to nil so that the mapper will get run again on the next loop
if ind == len(o) {
mapperOutputs[j] = nil
}
}
// if we didn't pull out any values, we're done here
if values == nil {
if !oneRowSent && !filterEmptyResults {
out <- m.processRawResults(nil)
}
return
}
// sort the values by time first so we can then handle offset and limit
sort.Sort(rawOutputs(values))
// get rid of any points that need to be offset
if valuesOffset < m.stmt.Offset {
offset := m.stmt.Offset - valuesOffset
// if offset is bigger than the number of values we have, move to the next batch from the mappers
if offset > len(values) {
valuesOffset += len(values)
continue
}
values = values[offset:]
valuesOffset += offset
}
// ensure we don't send more than the limit
if valuesSent < m.stmt.Limit {
limit := m.stmt.Limit - valuesSent
if len(values) > limit {
values = values[:limit]
}
valuesSent += len(values)
}
// convert the raw results into rows
row := m.processRawResults(values)
if filterEmptyResults && m.resultsEmpty(row.Values) {
return
}
// do any post processing like math and stuff
row.Values = m.processResults(row.Values)
oneRowSent = true
out <- row
// stop processing if we've hit the limit
if m.stmt.Limit != 0 && valuesSent >= m.stmt.Limit {
return
}
}
}
// processsResults will apply any math that was specified in the select statement against the passed in results
func (m *MapReduceJob) processResults(results [][]interface{}) [][]interface{} {
hasMath := false
for _, f := range m.stmt.Fields {
@ -245,8 +364,8 @@ func (m *MapReduceJob) processResults(results [][]interface{}) [][]interface{} {
// processFill will take the results and return new reaults (or the same if no fill modifications are needed) with whatever fill options the query has.
func (m *MapReduceJob) processFill(results [][]interface{}) [][]interface{} {
// don't do anything if it's raw query results or we're supposed to leave the nulls
if m.stmt.IsRawQuery || m.stmt.Fill == NullFill {
// don't do anything if we're supposed to leave the nulls
if m.stmt.Fill == NullFill {
return results
}
@ -396,6 +515,7 @@ func newBinaryExprEvaluator(op Token, lhs, rhs processor) processor {
}
}
// resultsEmpty will return true if the all the result values are empty or contain only nulls
func (m *MapReduceJob) resultsEmpty(resultValues [][]interface{}) bool {
for _, vals := range resultValues {
// start the loop at 1 because we want to skip over the time value
@ -409,7 +529,7 @@ func (m *MapReduceJob) resultsEmpty(resultValues [][]interface{}) bool {
}
// processRawResults will handle converting the reduce results from a raw query into a Row
func (m *MapReduceJob) processRawResults(resultValues [][]interface{}) *Row {
func (m *MapReduceJob) processRawResults(values []*rawQueryMapOutput) *Row {
selectNames := m.stmt.NamesInSelect()
// ensure that time is in the select names and in the first position
@ -440,14 +560,12 @@ func (m *MapReduceJob) processRawResults(resultValues [][]interface{}) *Row {
}
// return an empty row if there are no results
// resultValues should have exactly 1 array of interface. And for that array, the first element
// will be a time that we ignore, and the second element will be an array of []*rawMapOutput
if len(resultValues) == 0 || len(resultValues[0]) != ResultCountInRawResults {
if len(values) == 0 {
return row
}
// the results will have all of the raw mapper results, convert into the row
for _, v := range resultValues[0][1].([]*rawQueryMapOutput) {
for _, v := range values {
vals := make([]interface{}, len(selectNames))
if singleValue {
@ -468,21 +586,6 @@ func (m *MapReduceJob) processRawResults(resultValues [][]interface{}) *Row {
row.Values = append(row.Values, vals)
}
// apply limit and offset, if applicable
// TODO: make this so it doesn't read the whole result set into memory
if m.stmt.Limit > 0 || m.stmt.Offset > 0 {
if m.stmt.Offset > len(row.Values) {
row.Values = nil
} else {
limit := m.stmt.Limit
if m.stmt.Offset+m.stmt.Limit > len(row.Values) {
limit = len(row.Values) - m.stmt.Offset
}
row.Values = row.Values[m.stmt.Offset : m.stmt.Offset+limit]
}
}
return row
}
@ -496,10 +599,10 @@ func (m *MapReduceJob) processAggregate(c *Call, reduceFunc ReduceFunc, resultVa
}
}
firstInterval := m.interval
if !m.stmt.IsRawQuery {
firstInterval = (m.TMin/m.interval*m.interval + m.interval) - m.TMin
}
// the first interval in a query with a group by may be smaller than the others. This happens when they have a
// where time > clause that is in the middle of the bucket that the group by time creates
firstInterval := (m.TMin/m.interval*m.interval + m.interval) - m.TMin
// populate the result values for each interval of time
for i, _ := range resultValues {
// collect the results from each mapper
@ -542,9 +645,14 @@ type Mapper interface {
// NextInterval will get the time ordered next interval of the given interval size from the mapper. This is a
// forward only operation from the start time passed into Begin. Will return nil when there is no more data to be read.
// We pass the interval in here so that it can be varied over the period of the query. This is useful for the raw
// data queries where we'd like to gradually adjust the amount of time we scan over.
// We pass the interval in here so that it can be varied over the period of the query. This is useful for queries that
// must respect natural time boundaries like months or queries that span daylight savings time borders. Note that if
// a limit is set on the mapper, the interval passed here should represent the MaxTime in a nano epoch.
NextInterval(interval int64) (interface{}, error)
// Set limit will limit the number of data points yielded on the next interval. If a limit is set, the interval
// passed into NextInterval will be used as the MaxTime to scan until.
SetLimit(limit int)
}
type TagSet struct {
@ -616,6 +724,7 @@ func (p *Planner) Plan(stmt *SelectStatement, chunkSize int) (*Executor, error)
for _, j := range jobs {
j.interval = interval.Nanoseconds()
j.stmt = stmt
j.chunkSize = chunkSize
}
return &Executor{tx: tx, stmt: stmt, jobs: jobs, interval: interval.Nanoseconds()}, nil

22
influxql/functions.go
View File

@ -79,11 +79,6 @@ func InitializeMapFunc(c *Call) (MapFunc, error) {
// InitializeReduceFunc takes an aggregate call from the query and returns the ReduceFunc
func InitializeReduceFunc(c *Call) (ReduceFunc, error) {
// see if it's a query for raw data
if c == nil {
return ReduceRawQuery, nil
}
// Retrieve reduce function by name.
switch strings.ToLower(c.Name) {
case "count":
@ -515,7 +510,7 @@ func ReducePercentile(percentile float64) ReduceFunc {
// MapRawQuery is for queries without aggregates
func MapRawQuery(itr Iterator) interface{} {
var values []interface{}
var values []*rawQueryMapOutput
for _, k, v := itr.Next(); k != 0; _, k, v = itr.Next() {
val := &rawQueryMapOutput{k, v}
values = append(values, val)
@ -533,18 +528,3 @@ type rawOutputs []*rawQueryMapOutput
func (a rawOutputs) Len() int { return len(a) }
func (a rawOutputs) Less(i, j int) bool { return a[i].timestamp < a[j].timestamp }
func (a rawOutputs) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
// ReduceRawQuery is for queries without aggregates
func ReduceRawQuery(values []interface{}) interface{} {
allValues := make([]*rawQueryMapOutput, 0)
for _, v := range values {
if v == nil {
continue
}
for _, raw := range v.([]interface{}) {
allValues = append(allValues, raw.(*rawQueryMapOutput))
}
}
sort.Sort(rawOutputs(allValues))
return allValues
}

2
server.go
View File

@ -50,7 +50,7 @@ const (
// Defines the minimum duration allowed for all retention policies
retentionPolicyMinDuration = time.Hour
// When planning a select statement, passing zero tells it not to chunk results
// When planning a select statement, passing zero tells it not to chunk results. Only applies to raw queries
NoChunkingSize = 0
)

2
server_test.go
View File

@ -1946,7 +1946,7 @@ func (s *Server) MustWriteSeries(database, retentionPolicy string, points []infl
}
func (s *Server) executeQuery(q *influxql.Query, db string, user *influxdb.User) influxdb.Results {
results, err := s.ExecuteQuery(q, db, user, influxdb.NoChunkingSize)
results, err := s.ExecuteQuery(q, db, user, 10000)
if err != nil {
return influxdb.Results{Err: err}
}

27
tx.go
View File

@ -226,6 +226,7 @@ type LocalMapper struct {
selectFields []*Field // field names that occur in the select clause
selectTags []string // tag keys that occur in the select clause
isRaw bool // if the query is a non-aggregate query
limit int // used for raw queries to limit the amount of data read in before pushing out to client
}
// Open opens the LocalMapper.
@ -318,13 +319,16 @@ func (l *LocalMapper) Begin(c *influxql.Call, startingTime int64) error {
// NextInterval will get the time ordered next interval of the given interval size from the mapper. This is a
// forward only operation from the start time passed into Begin. Will return nil when there is no more data to be read.
// If this is a raw query, interval should be the max time to hit in the query
func (l *LocalMapper) NextInterval(interval int64) (interface{}, error) {
if l.cursorsEmpty || l.tmin > l.job.TMax {
return nil, nil
}
// Set the upper bound of the interval.
if interval > 0 {
if l.isRaw {
l.tmax = interval
} else if interval > 0 {
// Make sure the bottom of the interval lands on a natural boundary.
l.tmax = l.tmin + interval - 1
}
@ -341,15 +345,27 @@ func (l *LocalMapper) NextInterval(interval int64) (interface{}, error) {
}
}
// Move the interval forward.
l.tmin += interval
// Move the interval forward if it's not a raw query. For raw queries we use the limit to advance intervals.
if !l.isRaw {
l.tmin += interval
}
return val, nil
}
// SetLimit will tell the mapper to only yield that number of points (or to the max time) to Next
func (l *LocalMapper) SetLimit(limit int) {
l.limit = limit
}
// Next returns the next matching timestamped value for the LocalMapper.
func (l *LocalMapper) Next() (seriesID uint32, timestamp int64, value interface{}) {
for {
// if it's a raw query and we've hit the limit of the number of points to read in, bail
if l.isRaw && l.limit == 0 {
return uint32(0), int64(0), nil
}
// find the minimum timestamp
min := -1
minKey := int64(math.MaxInt64)
@ -415,6 +431,11 @@ func (l *LocalMapper) Next() (seriesID uint32, timestamp int64, value interface{
continue
}
// if it's a raw query, we always limit the amount we read in
if l.isRaw {
l.limit--
}
return seriesID, timestamp, value
}
}